[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 serves as a visibility-safe stop flag, the loop exits on interrupt, and the consumer is awaited before logsFilterPool shuts down so an in-flight capsule completes on graceful shutdown.
• ApplicationImpl.shutdown() now closes the consumer after producers stop and before dbManager.close(); 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). FullNode sets allowCircularReferences(false); @Lazy, like ObjectProvider or a runtime getBean(), only makes Spring tolerate the cycle. The reverse core → API edge stays, obscuring the component graph, and later json-rpc cleanups keep copying the pattern. This PR removes the edge instead.

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 the core Manager from JSON-RPC TronJsonRpcImpl by inserting a FilterCapsuleQueue between producers and the consumer. This removes the Spring circular reference, hardens shutdown (idempotent close, clean exit on interrupt, consumer stops before logsFilterPool), and keeps filter API behavior unchanged.

• Introduces FilterCapsuleQueue bean with a narrow API (offer, timed poll; stream() is test-only). Manager posts block/log filters to it, so it no longer references 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, warns on unknown FilterTriggerCapsule subtypes.
• Makes TronJsonRpcImpl.close() idempotent and stops the consumer before logsFilterPool; an interrupted closer returns while preserving the interrupt status and keeps logsFilterPool open until the consumer exits so an in-flight over-threshold capsule is not rejected. ApplicationImpl.shutdown() closes TronJsonRpcImpl after producers stop and before dbManager.close(), logging any close failures.
• Adds end-to-end delivery (full/solidity and reorg paths), consumer lifecycle (disabled mode, interrupt, interrupted closer, context destruction, repeated close), 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 3290dcb. 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: Advanced
• Run ID: b3e0c8bb-94df-42f1-8733-fbefe05d039f

• Autopilot · Keep fixing CodeRabbit findings and required CI, and resolving merge conflicts

---

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]

2 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/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]

All reported issues were addressed across 14 files

<sub>Requires human review: Auto-approval blocked because this review re-detected 1 unresolved issue already reported by Cubic.

Re-trigger cubic</sub>

[cubic-dev-ai]

All reported issues were addressed across 3 files (changes from recent commits).

<sub>Tip: Review your code locally with the cubic CLI to iterate faster.

Re-trigger cubic</sub>

[cubic-dev-ai]

3 issues found across 2 files (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java">

<violation number="1" location="framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java:1658">
P1: `close()` can shut down `logsFilterPool` while the filter consumer is still processing a capsule. Wait for `filterEs` to terminate before closing `logsFilterPool`; otherwise an interrupted shutdown can reject the consumer’s pending submission and lose that capsule.</violation>
</file>

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

<violation number="1" location="framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java:279">
P3: After closer.interrupt(), the test silently waits 500ms instead of verifying the comment's claim that close() is still blocked on the consumer. Assert closer.isAlive() here: it is deterministic while the consumer is blocked and fails immediately on the regression the test targets.</violation>

<violation number="2" location="framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java:286">
P3: This assertion message reads as if the rejected capsule is the expected outcome, but filters.isEmpty() expects the opposite: the pool outlived the consumer and processed the in-flight capsule. Reword it to state what must have happened for the assert to pass.</violation>
</file>

<sub>Tip: Review your code locally with the cubic CLI to iterate faster.

View guided diff | Turn on auto-fix | Re-trigger cubic</sub>

[cubic-dev-ai]

P1: close() can shut down logsFilterPool while the filter consumer is still processing a capsule. Wait for filterEs to terminate before closing logsFilterPool; otherwise an interrupted shutdown can reject the consumer’s pending submission and lose that capsule.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java, line 1658:

<comment>`close()` can shut down `logsFilterPool` while the filter consumer is still processing a capsule. Wait for `filterEs` to terminate before closing `logsFilterPool`; otherwise an interrupted shutdown can reject the consumer’s pending submission and lose that capsule.</comment>

<file context>
@@ -1651,8 +1652,12 @@ public void close() {
+    // terminate before the pool shuts down, even if the closing thread is interrupted.
     ExecutorServiceManager.shutdownAndAwaitTermination(filterEs, filterEsName);
+    if (filterEs != null && !filterEs.isTerminated()
+        && !Uninterruptibles.awaitTerminationUninterruptibly(filterEs, 2, TimeUnit.SECONDS)) {
+      logger.warn("Pool {} did not terminate before logs-filter-pool shutdown", filterEsName);
+    }
</file context>

[cubic-dev-ai]

P3: This assertion message reads as if the rejected capsule is the expected outcome, but filters.isEmpty() expects the opposite: the pool outlived the consumer and processed the in-flight capsule. Reword it to state what must have happened for the assert to pass.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java, line 286:

<comment>This assertion message reads as if the rejected capsule is the expected outcome, but filters.isEmpty() expects the opposite: the pool outlived the consumer and processed the in-flight capsule. Reword it to state what must have happened for the assert to pass.</comment>

<file context>
@@ -233,4 +241,48 @@ public boolean isSolidified() {
+    closer.join(GRACEFUL_BOUND_MS);
+    Assert.assertFalse("interrupted close() did not return", closer.isAlive());
+    Assert.assertTrue("consumer still alive", filterEs().awaitTermination(10, TimeUnit.SECONDS));
+    Assert.assertTrue("over-threshold capsule was rejected by a closed pool", filters.isEmpty());
+  }
 }
</file context>

Suggested change

	Assert.assertTrue("over-threshold capsule was rejected by a closed pool", filters.isEmpty());
	Assert.assertTrue("logsFilterPool shut down before the in-flight capsule was processed", filters.isEmpty());

[cubic-dev-ai]

P3: After closer.interrupt(), the test silently waits 500ms instead of verifying the comment's claim that close() is still blocked on the consumer. Assert closer.isAlive() here: it is deterministic while the consumer is blocked and fails immediately on the regression the test targets.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java, line 279:

<comment>After closer.interrupt(), the test silently waits 500ms instead of verifying the comment's claim that close() is still blocked on the consumer. Assert closer.isAlive() here: it is deterministic while the consumer is blocked and fails immediately on the regression the test targets.</comment>

<file context>
@@ -233,4 +241,48 @@ public boolean isSolidified() {
+      await(filterEs()::isShutdown, "close() did not start stopping the consumer");
+      closer.interrupt();
+      // without the fix close() returns here and shuts the pool before the release
+      closer.join(500);
+    } finally {
+      release.countDown();
</file context>

Suggested change

	closer.join(500);
	closer.join(500);
	Assert.assertTrue("close() returned while the consumer was blocked", closer.isAlive());