Skip to content

refactor(framework): decouple Manager from TronJsonRpcImpl #20

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
0xbigapple wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

refactor(framework): decouple Manager from TronJsonRpcImpl #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fe8d2a8
refactor(framework): decouple Manager from TronJsonRpcImpl
0xbigapple Aug 18, 2026
9f48952
fix(framework): exit filter loop on interrupt, widen shutdown catch
0xbigapple Aug 19, 2026
d9db070
refactor(jsonrpc): narrow filter queue api
0xbigapple Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions framework/src/main/java/org/tron/common/application/ApplicationImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import org.tron.core.db.Manager;
import org.tron.core.net.TronNetService;
import org.tron.core.services.event.EventService;
import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
import org.tron.program.SolidityNode;

@Slf4j(topic = "app")
Expand Down Expand Up @@ -37,6 +38,9 @@ public class ApplicationImpl implements Application {
@Autowired(required = false)
private SolidityNode solidityNode;

@Autowired
private TronJsonRpcImpl tronJsonRpc;

private final CountDownLatch shutdown = new CountDownLatch(1);

/**
Expand All @@ -62,6 +66,13 @@ public void shutdown() {
if (solidityNode != null) {
solidityNode.close();
}
// producers are stopped; stop the json-rpc filter consumer before the DB closes
// (idempotent — Spring bean destruction may call close() again)
try {
tronJsonRpc.close();
} catch (Exception e) {
logger.warn("Closing TronJsonRpcImpl failed.", e);
}
dbManager.close();
shutdown.countDown();
}
Expand Down
32 changes: 32 additions & 0 deletions framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
package org.tron.common.logsfilter.queue;

import com.google.common.annotations.VisibleForTesting;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.springframework.stereotype.Component;
import org.tron.common.logsfilter.capsule.FilterTriggerCapsule;

/**
* Queue between the block-processing producer (Manager) and the json-rpc filter
* consumer (TronJsonRpcImpl), so that neither side references the other.
*/
@Component
public class FilterCapsuleQueue {

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

@cubic-dev-ai cubic-dev-ai Bot Aug 19, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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);


public boolean offer(FilterTriggerCapsule capsule) {
return queue.offer(capsule);
}

public FilterTriggerCapsule poll(long timeout, TimeUnit unit) throws InterruptedException {
return queue.poll(timeout, unit);
}

@VisibleForTesting
public Stream<FilterTriggerCapsule> stream() {
return queue.stream();
}
}
55 changes: 5 additions & 50 deletions framework/src/main/java/org/tron/core/db/Manager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,6 @@
import org.apache.commons.collections4.CollectionUtils;
import org.bouncycastle.util.encoders.Hex;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import org.tron.api.GrpcAPI;
import org.tron.api.GrpcAPI.TransactionInfoList;
Expand All @@ -62,11 +61,11 @@
import org.tron.common.logsfilter.capsule.BlockFilterCapsule;
import org.tron.common.logsfilter.capsule.BlockLogTriggerCapsule;
import org.tron.common.logsfilter.capsule.ContractTriggerCapsule;
import org.tron.common.logsfilter.capsule.FilterTriggerCapsule;
import org.tron.common.logsfilter.capsule.LogsFilterCapsule;
import org.tron.common.logsfilter.capsule.SolidityTriggerCapsule;
import org.tron.common.logsfilter.capsule.TransactionLogTriggerCapsule;
import org.tron.common.logsfilter.capsule.TriggerCapsule;
import org.tron.common.logsfilter.queue.FilterCapsuleQueue;
import org.tron.common.logsfilter.trigger.ContractEventTrigger;
import org.tron.common.logsfilter.trigger.ContractLogTrigger;
import org.tron.common.logsfilter.trigger.ContractTrigger;
Expand Down Expand Up @@ -143,7 +142,6 @@
import org.tron.core.service.MortgageService;
import org.tron.core.service.RewardViCalService;
import org.tron.core.services.event.exception.EventException;
import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
import org.tron.core.store.AccountAssetStore;
import org.tron.core.store.AccountIdIndexStore;
import org.tron.core.store.AccountIndexStore;
Expand Down Expand Up @@ -253,8 +251,8 @@ public class Manager {
@Getter
private BlockingQueue<TriggerCapsule> triggerCapsuleQueue;
// log filter
private boolean isRunFilterProcessThread = true;
private BlockingQueue<FilterTriggerCapsule> filterCapsuleQueue;
@Autowired
private FilterCapsuleQueue filterCapsuleQueue;

@cubic-dev-ai cubic-dev-ai Bot Aug 19, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>


@Getter
private volatile long latestSolidityNumShutDown;
Expand All @@ -273,16 +271,10 @@ public class Manager {
private static final String rePushEsName = "repush";
private ExecutorService triggerEs;
private static final String triggerEsName = "event-trigger";
private ExecutorService filterEs;
private static final String filterEsName = "filter";

@Autowired
private RewardViCalService rewardViCalService;

@Lazy
@Autowired
private TronJsonRpcImpl tronJsonRpcImpl;

/**
* Cycle thread to rePush Transactions
*/
Expand Down Expand Up @@ -334,26 +326,6 @@ public class Manager {
}
};

private Runnable filterProcessLoop =
() -> {
while (isRunFilterProcessThread) {
try {
FilterTriggerCapsule filterCapsule = filterCapsuleQueue.poll(1, TimeUnit.SECONDS);
if (filterCapsule instanceof LogsFilterCapsule) {
tronJsonRpcImpl.handleLogsFilter((LogsFilterCapsule) filterCapsule);
} else if (filterCapsule instanceof BlockFilterCapsule) {
tronJsonRpcImpl.handleBLockFilter((BlockFilterCapsule) filterCapsule);
}
} catch (InterruptedException e) {
logger.error("FilterProcessLoop get InterruptedException, error is {}.",
e.getMessage());
Thread.currentThread().interrupt();
} catch (Throwable throwable) {
logger.error("Unknown throwable happened in filterProcessLoop. ", throwable);
}
}
};

private Comparator downComparator = (Comparator<TransactionCapsule>) (o1, o2) -> Long
.compare(o2.getOrder(), o1.getOrder());

Expand Down Expand Up @@ -476,11 +448,6 @@ public void stopRePushTriggerThread() {
ExecutorServiceManager.shutdownAndAwaitTermination(triggerEs, triggerEsName);
}

public void stopFilterProcessThread() {
isRunFilterProcessThread = false;
ExecutorServiceManager.shutdownAndAwaitTermination(filterEs, filterEsName);
}

public void stopValidateSignThread() {
ExecutorServiceManager.shutdownAndAwaitTermination(validateSignService, "validate-sign");
}
Expand Down Expand Up @@ -510,7 +477,6 @@ public void init() {
this.rePushTransactions = new LinkedBlockingQueue<>();
}
this.triggerCapsuleQueue = new LinkedBlockingQueue<>();
this.filterCapsuleQueue = new LinkedBlockingQueue<>();
chainBaseManager.setMerkleContainer(getMerkleContainer());
chainBaseManager.setMortgageService(mortgageService);
this.initGenesis();
Expand Down Expand Up @@ -584,12 +550,6 @@ public void init() {
ExecutorServiceManager.submit(triggerEs, triggerCapsuleProcessLoop);
}

// start json rpc filter process
if (CommonParameter.getInstance().isJsonRpcFilterEnabled()) {
filterEs = ExecutorServiceManager.newSingleThreadExecutor(filterEsName);
ExecutorServiceManager.submit(filterEs, filterProcessLoop);
}

//initStoreFactory
prepareStoreFactory();
//initActuatorCreator
Expand Down Expand Up @@ -2344,9 +2304,7 @@ private void reApplyBlockEvents(List<KhaosBlock> newBranch) {
private void postBlockFilter(final BlockCapsule blockCapsule, boolean solidified) {
BlockFilterCapsule blockFilterCapsule =
new BlockFilterCapsule(blockCapsule, solidified);
if (!filterCapsuleQueue.offer(blockFilterCapsule)) {
logger.info("Too many filters, block filter lost: {}.", blockCapsule.getBlockId());
}
filterCapsuleQueue.offer(blockFilterCapsule);
}

private void postLogsFilter(final BlockCapsule blockCapsule, boolean solidified,
Expand All @@ -2359,9 +2317,7 @@ private void postLogsFilter(final BlockCapsule blockCapsule, boolean solidified,
blockCapsule.getBlockId().toString(), blockCapsule.getBloom(), transactionInfoList,
solidified, removed);

if (!filterCapsuleQueue.offer(logsFilterCapsule)) {
logger.info("Too many filters, logs filter lost: {}.", blockNumber);
}
filterCapsuleQueue.offer(logsFilterCapsule);
}
}

Expand Down Expand Up @@ -2658,7 +2614,6 @@ public void close() {
stopRePushThread();
stopRePushTriggerThread();
EventPluginLoader.getInstance().stopPlugin();
stopFilterProcessThread();
stopValidateSignThread();
chainBaseManager.shutdown();
revokingStore.shutdown();
Expand Down
52 changes: 46 additions & 6 deletions framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,9 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;
import javax.annotation.PostConstruct;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
Expand All @@ -53,7 +55,9 @@
import org.tron.common.es.ExecutorServiceManager;
import org.tron.common.logsfilter.ContractEventParser;
import org.tron.common.logsfilter.capsule.BlockFilterCapsule;
import org.tron.common.logsfilter.capsule.FilterTriggerCapsule;
import org.tron.common.logsfilter.capsule.LogsFilterCapsule;
import org.tron.common.logsfilter.queue.FilterCapsuleQueue;
import org.tron.common.parameter.CommonParameter;
import org.tron.common.runtime.vm.DataWord;
import org.tron.common.utils.ByteArray;
Expand Down Expand Up @@ -193,20 +197,50 @@ public enum RequestSource {
private final ExecutorService sectionExecutor;
private final NodeInfoService nodeInfoService;
private final Wallet wallet;
@Autowired
private Manager manager;
private final Manager manager;
private final String esName = "query-section";

@Autowired
public TronJsonRpcImpl(@Autowired NodeInfoService nodeInfoService, @Autowired Wallet wallet) {
private FilterCapsuleQueue filterCapsuleQueue;
private ExecutorService filterEs;
private static final String filterEsName = "filter";
private final AtomicBoolean closed = new AtomicBoolean(false);

@Autowired
public TronJsonRpcImpl(NodeInfoService nodeInfoService, Wallet wallet, Manager manager) {
this.nodeInfoService = nodeInfoService;
this.wallet = wallet;
this.manager = manager;
this.sectionExecutor = ExecutorServiceManager.newFixedThreadPool(esName, 5);
}

@VisibleForTesting
public void setManager(Manager manager) {
this.manager = manager;
@PostConstruct
private void start() {
if (CommonParameter.getInstance().isJsonRpcFilterEnabled()) {
filterEs = ExecutorServiceManager.newSingleThreadExecutor(filterEsName, true);
ExecutorServiceManager.submit(filterEs, this::filterProcessLoop);
}
}

private void filterProcessLoop() {
while (!closed.get()) {
try {
FilterTriggerCapsule filterCapsule = filterCapsuleQueue.poll(1, TimeUnit.SECONDS);
if (filterCapsule instanceof LogsFilterCapsule) {
handleLogsFilter((LogsFilterCapsule) filterCapsule);
} else if (filterCapsule instanceof BlockFilterCapsule) {
handleBLockFilter((BlockFilterCapsule) filterCapsule);
} else if (filterCapsule != null) {
logger.warn("Unknown FilterTriggerCapsule: {}", filterCapsule.getClass().getName());
}
} catch (InterruptedException e) {
logger.error("FilterProcessLoop get InterruptedException, error is {}.", e.getMessage());
Thread.currentThread().interrupt();
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
return;
} catch (Throwable throwable) {
logger.error("Unknown throwable happened in filterProcessLoop. ", throwable);
}
}
}

@VisibleForTesting
Expand Down Expand Up @@ -1614,6 +1648,12 @@ public Object[] getFilterResult(String filterId, Map<String, BlockFilterAndResul

@Override
public void close() throws IOException {
if (!closed.compareAndSet(false, true)) {
return;
}
// The consumer loop submits to logsFilterPool (over-threshold path), so it must
// terminate before the pool shuts down.
ExecutorServiceManager.shutdownAndAwaitTermination(filterEs, filterEsName);
ExecutorServiceManager.shutdownAndAwaitTermination(logsFilterPool, "logs-filter-pool");
logElementCache.invalidateAll();
blockHashCache.invalidateAll();
Expand Down
3 changes: 1 addition & 2 deletions framework/src/test/java/org/tron/common/runtime/vm/Create2Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,8 +209,7 @@ private void testJsonRpc(byte[] actualContract, long loop) {
NodeInfoService nodeInfoService;
nodeInfoService = context.getBean(NodeInfoService.class);
Wallet wallet = context.getBean(Wallet.class);
tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet);
tronJsonRpc.setManager(manager);
tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, manager);
try {
String res =
tronJsonRpc.getStorageAt(ByteArray.toHexString(actualContract), "0", "latest");
Expand Down
14 changes: 6 additions & 8 deletions framework/src/test/java/org/tron/core/db/ManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,10 +45,10 @@
import org.tron.common.logsfilter.EventPluginLoader;
import org.tron.common.logsfilter.capsule.BlockFilterCapsule;
import org.tron.common.logsfilter.capsule.BlockLogTriggerCapsule;
import org.tron.common.logsfilter.capsule.FilterTriggerCapsule;
import org.tron.common.logsfilter.capsule.LogsFilterCapsule;
import org.tron.common.logsfilter.capsule.TransactionLogTriggerCapsule;
import org.tron.common.logsfilter.capsule.TriggerCapsule;
import org.tron.common.logsfilter.queue.FilterCapsuleQueue;
import org.tron.common.logsfilter.trigger.ContractLogTrigger;
import org.tron.common.parameter.CommonParameter;
import org.tron.common.runtime.RuntimeImpl;
Expand Down Expand Up @@ -1694,8 +1694,8 @@ public void adjustBalance(AccountStore accountStore, byte[] accountAddress, long
@Test
public void switchForkShouldPostFullNodeFilterForNewBranch() throws Exception {
CommonParameter.getInstance().jsonRpcHttpFullNodeEnable = true;
// filterProcessLoop only starts when isJsonRpcFilterEnabled() held at Manager.init() time; it
// was false then, so filterCapsuleQueue is produce-only here and fully observable.
// The consumer thread only starts when isJsonRpcFilterEnabled() held at context startup; it
// was false then, so the FilterCapsuleQueue bean is produce-only here and fully observable.

// bootstrap a head with a known witness
String key = PublicMethod.getRandomPrivateKey();
Expand Down Expand Up @@ -1733,9 +1733,7 @@ public void switchForkShouldPostFullNodeFilterForNewBranch() throws Exception {
dbManager.pushBlock(p);

long expiration = t + 1_000_000L;
BlockingQueue<FilterTriggerCapsule> queue =
ReflectUtils.getFieldValue(dbManager, "filterCapsuleQueue");
queue.clear();
FilterCapsuleQueue queue = context.getBean(FilterCapsuleQueue.class);

// old branch: A carries a transfer; applied via the normal extend path
BlockCapsule a = blockWithTransfer(t + 6000, base + 2, p.getBlockId().getByteString(), keys,
Expand Down Expand Up @@ -1868,7 +1866,7 @@ private BlockCapsule blockWithTransfer(long time, long number, ByteString parent
return blockCapsule;
}

private boolean hasLogsFilterCapsule(BlockingQueue<FilterTriggerCapsule> queue, BlockCapsule b,
private boolean hasLogsFilterCapsule(FilterCapsuleQueue queue, BlockCapsule b,
boolean removed) {
String blockHash = b.getBlockId().toString();
return queue.stream()
Expand All @@ -1878,7 +1876,7 @@ private boolean hasLogsFilterCapsule(BlockingQueue<FilterTriggerCapsule> queue,
&& blockHash.equals(c.getBlockHash()));
}

private boolean hasBlockFilterCapsule(BlockingQueue<FilterTriggerCapsule> queue,
private boolean hasBlockFilterCapsule(FilterCapsuleQueue queue,
BlockCapsule b) {
String blockHash = b.getBlockId().toString();
return queue.stream()
Expand Down
2 changes: 1 addition & 1 deletion framework/src/test/java/org/tron/core/jsonrpc/ConcurrentHashMapTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
@Slf4j
public class ConcurrentHashMapTest {
private static final String EXECUTOR_NAME = "jsonrpc-concurrent-map-test";
private final TronJsonRpcImpl jsonRpc = new TronJsonRpcImpl(null, null);
private final TronJsonRpcImpl jsonRpc = new TronJsonRpcImpl(null, null, null);

private static int randomInt(int minInt, int maxInt) {
return (int) round(random(true) * (maxInt - minInt) + minInt, true);
Expand Down
Loading
Loading