Repository navigation
refactor(framework): decouple Manager from TronJsonRpcImpl - #20
0xbigapple wants to merge 8 commits into
Conversation
|
Important
This repository does not receive automatic reviews because it has fewer than 10 stars. ⚙️ Run configuration
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. Comment |
There was a problem hiding this comment.
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>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| @Component | ||
| public class FilterCapsuleQueue { | ||
|
|
||
| private final BlockingQueue<FilterTriggerCapsule> queue = new LinkedBlockingQueue<>(); |
There was a problem hiding this comment.
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>
| private final BlockingQueue<FilterTriggerCapsule> queue = new LinkedBlockingQueue<>(); | |
| private final BlockingQueue<FilterTriggerCapsule> queue = new LinkedBlockingQueue<>(10000); |
| private boolean isRunFilterProcessThread = true; | ||
| private BlockingQueue<FilterTriggerCapsule> filterCapsuleQueue; | ||
| @Autowired | ||
| private FilterCapsuleQueue filterCapsuleQueue; |
There was a problem hiding this comment.
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>
2380bd9 to
9f48952
Compare
Remove unused queue inspection and mutation methods, and mark the remaining stream accessor as test-only. Drop unreachable offer failure handling for the unbounded queue.
There was a problem hiding this comment.
All reported issues were addressed across 14 files
Requires human review: Auto-approval blocked because this review re-detected 1 unresolved issue already reported by Cubic.
Re-trigger cubic
There was a problem hiding this comment.
All reported issues were addressed across 3 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
e506969 to
f51e02a
Compare
There was a problem hiding this comment.
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>
Tip: Review your code locally with the cubic CLI to iterate faster.
View guided diff | Turn on auto-fix | Re-trigger cubic
| // 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)) { |
There was a problem hiding this comment.
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>
| 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()); |
There was a problem hiding this comment.
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>
| 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()); |
| 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); |
There was a problem hiding this comment.
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>
| closer.join(500); | |
| closer.join(500); | |
| Assert.assertTrue("close() returned while the consumer was blocked", closer.isAlive()); |
What does this PR do?
Removes the
Manager → TronJsonRpcImplreverse dependency left by tronprotocol#6732, where core-layerManagerholds the json-rpc filter consumer through a@Lazyinjection.FilterCapsuleQueuebean;Manager.postBlockFilter/postLogsFilterproduce into it, so Manager no longer references anything underorg.tron.core.services.jsonrpc(repo-wideframework/src/mainis now@Lazy-free).TronJsonRpcImpl: started by@PostConstructwhenisJsonRpcFilterEnabled(), single daemon thread, stopped byclose().TronJsonRpcImplconstructor becomes(NodeInfoService, Wallet, Manager);setManager()is removed and the direct-construction test sites are migrated.close()is rewritten: anAtomicBooleanguard makes it idempotent and serves as a visibility-safe stop flag, the loop exits on interrupt, and the consumer is awaited beforelogsFilterPoolshuts down so an in-flight capsule completes on graceful shutdown.ApplicationImpl.shutdown()now closes the consumer after producers stop and beforedbManager.close(); the later Spring-destructionclose()is a no-op.instanceofdispatch logs a warning for unknownFilterTriggerCapsulesubtypes 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).
FullNodesetsallowCircularReferences(false);@Lazy, likeObjectProvideror a runtimegetBean(), 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:
eth_getFilterChangesdelivers; SIGTERM log order confirms consumer →logs-filter-pool→dbManager→ context close, single shutdown episode, no errorsfilter not foundafter restart; noRejectedExecutionException, no bean-cycle errors, shutdown order identical in every cycleFollow up
Extra details
Summary by cubic
Decouples the core
Managerfrom JSON-RPCTronJsonRpcImplby inserting aFilterCapsuleQueuebetween producers and the consumer. This removes the Spring circular reference, hardens shutdown (idempotent close, clean exit on interrupt, consumer stops beforelogsFilterPool), and keeps filter API behavior unchanged.FilterCapsuleQueuebean with a narrow API (offer, timedpoll;stream()is test-only).Managerposts block/log filters to it, so it no longer referencesorg.tron.core.services.jsonrpc. Drops unreachable offer-failure handling since the queue is unbounded.TronJsonRpcImpl; starts at@PostConstructonly whenisJsonRpcFilterEnabled(), runs in a single daemon thread, exits on interrupt, warns on unknownFilterTriggerCapsulesubtypes.TronJsonRpcImpl.close()idempotent and stops the consumer beforelogsFilterPool; an interrupted closer returns while preserving the interrupt status and keepslogsFilterPoolopen until the consumer exits so an in-flight over-threshold capsule is not rejected.ApplicationImpl.shutdown()closesTronJsonRpcImplafter producers stop and beforedbManager.close(), logging any close failures.Migration
new TronJsonRpcImpl(nodeInfoService, wallet, manager)and removesetManager(...)calls.FilterCapsuleQueueand useoffer(...); for tests, use itsstream()accessor.Written for commit 3290dcb. Summary will update on new commits.