From fe8d2a8602cd85568f9bb0eed22b1922ed740dd5 Mon Sep 17 00:00:00 2001 From: 0xbigapple Date: Tue, 18 Aug 2026 14:31:27 +0800 Subject: [PATCH 1/8] refactor(framework): decouple Manager from TronJsonRpcImpl --- .../common/application/ApplicationImpl.java | 12 ++ .../logsfilter/queue/FilterCapsuleQueue.java | 38 ++++++ .../main/java/org/tron/core/db/Manager.java | 47 +------ .../services/jsonrpc/TronJsonRpcImpl.java | 51 +++++++- .../tron/common/runtime/vm/Create2Test.java | 3 +- .../java/org/tron/core/db/ManagerTest.java | 13 +- .../core/jsonrpc/ConcurrentHashMapTest.java | 2 +- .../jsonrpc/FilterPipelineDeliveryTest.java | 100 +++++++++++++++ .../jsonrpc/FilterPipelineShutdownTest.java | 120 ++++++++++++++++++ .../core/jsonrpc/HandleLogsFilterTest.java | 2 +- .../JsonRpcCallAndEstimateGasTest.java | 9 +- .../tron/core/jsonrpc/JsonrpcServiceTest.java | 3 +- .../tron/core/jsonrpc/WalletCursorTest.java | 15 +-- 13 files changed, 336 insertions(+), 79 deletions(-) create mode 100644 framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java create mode 100644 framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java create mode 100644 framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java diff --git a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java index bab95d299ab..f1bba10937a 100644 --- a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java +++ b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java @@ -1,5 +1,6 @@ package org.tron.common.application; +import java.io.IOException; import java.util.concurrent.CountDownLatch; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -10,6 +11,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") @@ -37,6 +39,9 @@ public class ApplicationImpl implements Application { @Autowired(required = false) private SolidityNode solidityNode; + @Autowired + private TronJsonRpcImpl tronJsonRpc; + private final CountDownLatch shutdown = new CountDownLatch(1); /** @@ -62,6 +67,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 (IOException e) { + logger.warn("Closing TronJsonRpcImpl failed.", e); + } dbManager.close(); shutdown.countDown(); } diff --git a/framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java b/framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java new file mode 100644 index 00000000000..7b5ba0e550d --- /dev/null +++ b/framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java @@ -0,0 +1,38 @@ +package org.tron.common.logsfilter.queue; + +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 queue = new LinkedBlockingQueue<>(); + + public boolean offer(FilterTriggerCapsule capsule) { + return queue.offer(capsule); + } + + public FilterTriggerCapsule poll(long timeout, TimeUnit unit) throws InterruptedException { + return queue.poll(timeout, unit); + } + + public int size() { + return queue.size(); + } + + public void clear() { + queue.clear(); + } + + public Stream stream() { + return queue.stream(); + } +} diff --git a/framework/src/main/java/org/tron/core/db/Manager.java b/framework/src/main/java/org/tron/core/db/Manager.java index 9d7a7c979b9..162325573dd 100644 --- a/framework/src/main/java/org/tron/core/db/Manager.java +++ b/framework/src/main/java/org/tron/core/db/Manager.java @@ -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; @@ -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; @@ -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; @@ -253,8 +251,8 @@ public class Manager { @Getter private BlockingQueue triggerCapsuleQueue; // log filter - private boolean isRunFilterProcessThread = true; - private BlockingQueue filterCapsuleQueue; + @Autowired + private FilterCapsuleQueue filterCapsuleQueue; @Getter private volatile long latestSolidityNumShutDown; @@ -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 */ @@ -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) (o1, o2) -> Long .compare(o2.getOrder(), o1.getOrder()); @@ -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"); } @@ -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(); @@ -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 @@ -2658,7 +2618,6 @@ public void close() { stopRePushThread(); stopRePushTriggerThread(); EventPluginLoader.getInstance().stopPlugin(); - stopFilterProcessThread(); stopValidateSignThread(); chainBaseManager.shutdown(); revokingStore.shutdown(); diff --git a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java index 6be47886117..44deee06859 100644 --- a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java +++ b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java @@ -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; @@ -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; @@ -193,20 +197,49 @@ 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(); + } catch (Throwable throwable) { + logger.error("Unknown throwable happened in filterProcessLoop. ", throwable); + } + } } @VisibleForTesting @@ -1614,6 +1647,12 @@ public Object[] getFilterResult(String filterId, Map queue = - ReflectUtils.getFieldValue(dbManager, "filterCapsuleQueue"); + FilterCapsuleQueue queue = context.getBean(FilterCapsuleQueue.class); queue.clear(); // old branch: A carries a transfer; applied via the normal extend path @@ -1868,7 +1867,7 @@ private BlockCapsule blockWithTransfer(long time, long number, ByteString parent return blockCapsule; } - private boolean hasLogsFilterCapsule(BlockingQueue queue, BlockCapsule b, + private boolean hasLogsFilterCapsule(FilterCapsuleQueue queue, BlockCapsule b, boolean removed) { String blockHash = b.getBlockId().toString(); return queue.stream() @@ -1878,7 +1877,7 @@ private boolean hasLogsFilterCapsule(BlockingQueue queue, && blockHash.equals(c.getBlockHash())); } - private boolean hasBlockFilterCapsule(BlockingQueue queue, + private boolean hasBlockFilterCapsule(FilterCapsuleQueue queue, BlockCapsule b) { String blockHash = b.getBlockId().toString(); return queue.stream() diff --git a/framework/src/test/java/org/tron/core/jsonrpc/ConcurrentHashMapTest.java b/framework/src/test/java/org/tron/core/jsonrpc/ConcurrentHashMapTest.java index 2fcb624002e..855aae3f22d 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/ConcurrentHashMapTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/ConcurrentHashMapTest.java @@ -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); diff --git a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java new file mode 100644 index 00000000000..97bd4e03c21 --- /dev/null +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java @@ -0,0 +1,100 @@ +package org.tron.core.jsonrpc; + +import java.util.Collections; +import java.util.List; +import java.util.function.BooleanSupplier; +import javax.annotation.Resource; +import org.junit.Assert; +import org.junit.Test; +import org.tron.common.BaseTest; +import org.tron.common.TestConstants; +import org.tron.common.logsfilter.capsule.BlockFilterCapsule; +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.runtime.vm.LogInfo; +import org.tron.core.config.args.Args; +import org.tron.core.services.jsonrpc.TronJsonRpc.FilterRequest; +import org.tron.core.services.jsonrpc.TronJsonRpc.LogFilterElement; +import org.tron.core.services.jsonrpc.TronJsonRpcImpl; +import org.tron.core.services.jsonrpc.filters.BlockFilterAndResult; +import org.tron.core.services.jsonrpc.filters.LogFilterAndResult; +import org.tron.protos.Protocol.TransactionInfo; + +/** + * End-to-end coverage of the decoupled filter pipeline: capsules offered to the + * FilterCapsuleQueue bean are delivered to registered filters by the consumer + * thread that TronJsonRpcImpl starts at context startup. + */ +public class FilterPipelineDeliveryTest extends BaseTest { + + static { + Args.setParam(new String[] {"--output-directory", dbPath()}, TestConstants.TEST_CONF); + // isJsonRpcFilterEnabled() must hold at context startup so the consumer thread starts + CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(true); + } + + @Resource + private TronJsonRpcImpl tronJsonRpc; + @Resource + private FilterCapsuleQueue filterCapsuleQueue; + + private static TransactionInfo buildTxInfoWithLog() { + LogInfo logInfo = new LogInfo(new byte[20], + Collections.singletonList(new DataWord(new byte[32])), new byte[0]); + return TransactionInfo.newBuilder().addLog(LogInfo.buildLog(logInfo)).build(); + } + + private static void await(BooleanSupplier condition, String message) + throws InterruptedException { + long deadline = System.currentTimeMillis() + 10_000; + while (System.currentTimeMillis() < deadline) { + if (condition.getAsBoolean()) { + return; + } + Thread.sleep(50); + } + Assert.fail(message); + } + + @Test + public void consumerThreadStartedOnce() { + long count = Thread.getAllStackTraces().keySet().stream() + .filter(t -> "filter".equals(t.getName())).count(); + Assert.assertEquals(1, count); + } + + @Test + public void blockFilterDeliveredOnFullAndSolidityPaths() throws Exception { + BlockFilterAndResult full = new BlockFilterAndResult(); + BlockFilterAndResult solidity = new BlockFilterAndResult(); + tronJsonRpc.getBlockFilter2ResultFull().put("pipeline-block-full", full); + tronJsonRpc.getBlockFilter2ResultSolidity().put("pipeline-block-solidity", solidity); + try { + filterCapsuleQueue.offer(new BlockFilterCapsule("e2e-full-hash", false)); + filterCapsuleQueue.offer(new BlockFilterCapsule("e2e-solidity-hash", true)); + await(() -> full.getResult().size() == 1, "full-path block hash not delivered"); + await(() -> solidity.getResult().size() == 1, "solidity-path block hash not delivered"); + } finally { + tronJsonRpc.getBlockFilter2ResultFull().remove("pipeline-block-full"); + tronJsonRpc.getBlockFilter2ResultSolidity().remove("pipeline-block-solidity"); + } + } + + @Test + public void reorgLogsDeliveredWithRemovedFlag() throws Exception { + LogFilterAndResult filter = new LogFilterAndResult(new FilterRequest(), 100L, null); + tronJsonRpc.getEventFilter2ResultFull().put("pipeline-log-removed", filter); + try { + filterCapsuleQueue.offer(new LogsFilterCapsule(150L, "0xreorg", null, + Collections.singletonList(buildTxInfoWithLog()), false, true)); + await(() -> !filter.getResult().isEmpty(), "removed log not delivered"); + List elements = filter.popAll(); + Assert.assertEquals(1, elements.size()); + Assert.assertTrue(elements.get(0).isRemoved()); + } finally { + tronJsonRpc.getEventFilter2ResultFull().remove("pipeline-log-removed"); + } + } +} diff --git a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java new file mode 100644 index 00000000000..b58e95a5784 --- /dev/null +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java @@ -0,0 +1,120 @@ +package org.tron.core.jsonrpc; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import javax.annotation.Resource; +import org.junit.Assert; +import org.junit.FixMethodOrder; +import org.junit.Test; +import org.junit.runners.MethodSorters; +import org.tron.common.BaseTest; +import org.tron.common.TestConstants; +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.runtime.vm.LogInfo; +import org.tron.core.config.args.Args; +import org.tron.core.services.jsonrpc.TronJsonRpc.FilterRequest; +import org.tron.core.services.jsonrpc.TronJsonRpcImpl; +import org.tron.core.services.jsonrpc.filters.LogFilterAndResult; +import org.tron.protos.Protocol.TransactionInfo; + +/** + * Shutdown semantics of the decoupled filter pipeline. Method order matters: + * test1 closes the shared TronJsonRpcImpl bean, test2 asserts the second close + * is a guarded no-op, so they run in name order. + */ +@FixMethodOrder(MethodSorters.NAME_ASCENDING) +public class FilterPipelineShutdownTest extends BaseTest { + + static { + Args.setParam(new String[] {"--output-directory", dbPath()}, TestConstants.TEST_CONF); + // isJsonRpcFilterEnabled() must hold at context startup so the consumer thread starts + CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(true); + } + + @Resource + private TronJsonRpcImpl tronJsonRpc; + @Resource + private FilterCapsuleQueue filterCapsuleQueue; + + private static TransactionInfo buildTxInfoWithLog() { + LogInfo logInfo = new LogInfo(new byte[20], + Collections.singletonList(new DataWord(new byte[32])), new byte[0]); + return TransactionInfo.newBuilder().addLog(LogInfo.buildLog(logInfo)).build(); + } + + /** + * close() while the consumer is mid-capsule on the parallel (logsFilterPool) path: + * the in-flight capsule must complete instead of dying on a RejectedExecutionException, + * which is exactly the filterEs-before-logsFilterPool ordering constraint. + */ + @Test + public void test1CloseDuringParallelProcessingLosesNoEvent() throws Exception { + tronJsonRpc.setFilterParallelThreshold(0); + LogFilterAndResult filter = new LogFilterAndResult(new FilterRequest(), 100L, null); + tronJsonRpc.getEventFilter2ResultFull().put("shutdown-race", filter); + + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + LogsFilterCapsule capsule = new LogsFilterCapsule(150L, "0xrace", null, + Collections.singletonList(buildTxInfoWithLog()), false, false) { + @Override + public boolean isSolidified() { + // first call is the head of handleLogsFilter, on the consumer thread + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return false; + } + }; + filterCapsuleQueue.offer(capsule); + Assert.assertTrue("consumer did not pick up the capsule", + entered.await(10, TimeUnit.SECONDS)); + + Thread closer = new Thread(() -> { + try { + tronJsonRpc.close(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }, "test-closer"); + closer.start(); + // close() must be parked awaiting the consumer, with logsFilterPool still open + Thread.sleep(500); + Assert.assertTrue("close() finished while a capsule was in flight", closer.isAlive()); + + release.countDown(); + closer.join(150_000); + Assert.assertFalse("close() did not finish", closer.isAlive()); + Assert.assertEquals("in-flight capsule was lost during close()", + 1, filter.getResult().size()); + } + + @Test + public void test2RepeatedCloseIsIdempotentAndThreadStopped() throws Exception { + long deadline = System.currentTimeMillis() + 10_000; + while (System.currentTimeMillis() < deadline && filterThreadAlive()) { + Thread.sleep(50); + } + Assert.assertFalse("consumer thread still alive after close()", filterThreadAlive()); + + long t0 = System.nanoTime(); + tronJsonRpc.close(); + long elapsedMs = (System.nanoTime() - t0) / 1_000_000; + Assert.assertTrue("repeated close should be a fast no-op, took " + elapsedMs + "ms", + elapsedMs < 1_000); + } + + private static boolean filterThreadAlive() { + return Thread.getAllStackTraces().keySet().stream() + .anyMatch(t -> "filter".equals(t.getName()) && t.isAlive()); + } +} diff --git a/framework/src/test/java/org/tron/core/jsonrpc/HandleLogsFilterTest.java b/framework/src/test/java/org/tron/core/jsonrpc/HandleLogsFilterTest.java index 33835c482fe..be3098095d5 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/HandleLogsFilterTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/HandleLogsFilterTest.java @@ -27,7 +27,7 @@ public class HandleLogsFilterTest { @Before public void setUp() { - jsonRpc = new TronJsonRpcImpl(null, null); + jsonRpc = new TronJsonRpcImpl(null, null, null); } @After diff --git a/framework/src/test/java/org/tron/core/jsonrpc/JsonRpcCallAndEstimateGasTest.java b/framework/src/test/java/org/tron/core/jsonrpc/JsonRpcCallAndEstimateGasTest.java index 2ab455fa580..cc11b2e99f1 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/JsonRpcCallAndEstimateGasTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/JsonRpcCallAndEstimateGasTest.java @@ -211,8 +211,7 @@ private static TronJsonRpcImpl newRpcWithMockedFailedCall(byte[] resData, Estima }); } - TronJsonRpcImpl rpc = new TronJsonRpcImpl(mockNodeInfo, mockWallet); - rpc.setManager(mockManager); + TronJsonRpcImpl rpc = new TronJsonRpcImpl(mockNodeInfo, mockWallet, mockManager); return rpc; } @@ -239,8 +238,7 @@ private static TronJsonRpcImpl newRpcWithMockedSuccessfulCall(byte[]... constant .build(); }); - TronJsonRpcImpl rpc = new TronJsonRpcImpl(mockNodeInfo, mockWallet); - rpc.setManager(mockManager); + TronJsonRpcImpl rpc = new TronJsonRpcImpl(mockNodeInfo, mockWallet, mockManager); return rpc; } @@ -276,8 +274,7 @@ private static TronJsonRpcImpl newRpcWithMockedEstimateGasSuccessfulCall(long en }); } - TronJsonRpcImpl rpc = new TronJsonRpcImpl(mockNodeInfo, mockWallet); - rpc.setManager(mockManager); + TronJsonRpcImpl rpc = new TronJsonRpcImpl(mockNodeInfo, mockWallet, mockManager); return rpc; } } diff --git a/framework/src/test/java/org/tron/core/jsonrpc/JsonrpcServiceTest.java b/framework/src/test/java/org/tron/core/jsonrpc/JsonrpcServiceTest.java index e8d14ace060..a564396198d 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/JsonrpcServiceTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/JsonrpcServiceTest.java @@ -216,8 +216,7 @@ public void init() { dbManager.getTransactionRetStore() .put(ByteArray.fromLong(blockCapsule2.getNum()), transactionRetCapsule2); - tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet); - tronJsonRpc.setManager(dbManager); + tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, dbManager); } @Test diff --git a/framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java b/framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java index 24ca71a74bc..c26d81e1f99 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java @@ -60,8 +60,7 @@ public void init() { @Test public void testSource() { - TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet); - tronJsonRpc.setManager(dbManager); + TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, dbManager); Assert.assertEquals(Cursor.HEAD, wallet.getCursor()); Assert.assertEquals(RequestSource.FULLNODE, tronJsonRpc.getSource()); @@ -92,8 +91,7 @@ public void testDisableInSolidity() { dbManager.setCursor(Cursor.SOLIDITY); - TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet); - tronJsonRpc.setManager(dbManager); + TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, dbManager); try { tronJsonRpc.buildTransaction(buildArguments); tronJsonRpc.close(); @@ -115,8 +113,7 @@ public void testDisableInPBFT() { dbManager.setCursor(Cursor.PBFT); - TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet); - tronJsonRpc.setManager(dbManager); + TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, dbManager); try { tronJsonRpc.buildTransaction(buildArguments); } catch (Exception e) { @@ -143,8 +140,7 @@ public void testEnableInFullNode() { buildArguments.setTo("0x548794500882809695a8a687866e76d4271a1abc"); buildArguments.setValue("0x1f4"); - TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet); - tronJsonRpc.setManager(dbManager); + TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, dbManager); try { tronJsonRpc.buildTransaction(buildArguments); @@ -164,8 +160,7 @@ public void testNewFilter_exceedsCapThrowsException() throws Exception { int saved = Args.getInstance().getJsonRpcMaxLogFilterNum(); Args.getInstance().setJsonRpcMaxLogFilterNum(cap); FilterRequest fr = new FilterRequest(); - TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet); - tronJsonRpc.setManager(dbManager); + TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, dbManager); Map map = tronJsonRpc.getEventFilter2ResultFull(); List addedKeys = new ArrayList<>(); From 9f4895267af217560382a1c8e0b115e31f8ad5ec Mon Sep 17 00:00:00 2001 From: 0xbigapple Date: Wed, 19 Aug 2026 10:07:36 +0800 Subject: [PATCH 2/8] fix(framework): exit filter loop on interrupt, widen shutdown catch --- .../main/java/org/tron/common/application/ApplicationImpl.java | 3 +-- .../java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java | 1 + 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java index f1bba10937a..88c15ae26fc 100644 --- a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java +++ b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java @@ -1,6 +1,5 @@ package org.tron.common.application; -import java.io.IOException; import java.util.concurrent.CountDownLatch; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -71,7 +70,7 @@ public void shutdown() { // (idempotent — Spring bean destruction may call close() again) try { tronJsonRpc.close(); - } catch (IOException e) { + } catch (Exception e) { logger.warn("Closing TronJsonRpcImpl failed.", e); } dbManager.close(); diff --git a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java index 44deee06859..890ffb9081a 100644 --- a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java +++ b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java @@ -236,6 +236,7 @@ private void filterProcessLoop() { } catch (InterruptedException e) { logger.error("FilterProcessLoop get InterruptedException, error is {}.", e.getMessage()); Thread.currentThread().interrupt(); + return; } catch (Throwable throwable) { logger.error("Unknown throwable happened in filterProcessLoop. ", throwable); } From d9db07054c4701a5c23754a377cd5c7f8967dc55 Mon Sep 17 00:00:00 2001 From: 0xbigapple Date: Wed, 19 Aug 2026 21:18:29 +0800 Subject: [PATCH 3/8] refactor(jsonrpc): narrow filter queue api 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. --- .../common/logsfilter/queue/FilterCapsuleQueue.java | 10 ++-------- framework/src/main/java/org/tron/core/db/Manager.java | 8 ++------ .../src/test/java/org/tron/core/db/ManagerTest.java | 1 - 3 files changed, 4 insertions(+), 15 deletions(-) diff --git a/framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java b/framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java index 7b5ba0e550d..35f551966f1 100644 --- a/framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java +++ b/framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java @@ -1,5 +1,6 @@ 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; @@ -24,14 +25,7 @@ public FilterTriggerCapsule poll(long timeout, TimeUnit unit) throws Interrupted return queue.poll(timeout, unit); } - public int size() { - return queue.size(); - } - - public void clear() { - queue.clear(); - } - + @VisibleForTesting public Stream stream() { return queue.stream(); } diff --git a/framework/src/main/java/org/tron/core/db/Manager.java b/framework/src/main/java/org/tron/core/db/Manager.java index 162325573dd..d4dbddf2768 100644 --- a/framework/src/main/java/org/tron/core/db/Manager.java +++ b/framework/src/main/java/org/tron/core/db/Manager.java @@ -2304,9 +2304,7 @@ private void reApplyBlockEvents(List 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, @@ -2319,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); } } diff --git a/framework/src/test/java/org/tron/core/db/ManagerTest.java b/framework/src/test/java/org/tron/core/db/ManagerTest.java index 5a973889ec2..5a903925f19 100755 --- a/framework/src/test/java/org/tron/core/db/ManagerTest.java +++ b/framework/src/test/java/org/tron/core/db/ManagerTest.java @@ -1734,7 +1734,6 @@ public void switchForkShouldPostFullNodeFilterForNewBranch() throws Exception { long expiration = t + 1_000_000L; FilterCapsuleQueue queue = context.getBean(FilterCapsuleQueue.class); - queue.clear(); // old branch: A carries a transfer; applied via the normal extend path BlockCapsule a = blockWithTransfer(t + 6000, base + 2, p.getBlockId().getByteString(), keys, From f51e02a53161c0e515702a24a05c6fcb9940a014 Mon Sep 17 00:00:00 2001 From: 0xbigapple Date: Tue, 22 Sep 2026 10:55:55 +0800 Subject: [PATCH 4/8] test(jsonrpc): cover filter consumer lifecycle and tighten filter pipeline tests --- .../jsonrpc/FilterPipelineDeliveryTest.java | 10 +- .../jsonrpc/FilterPipelineLifecycleTest.java | 237 ++++++++++++++++++ .../jsonrpc/FilterPipelineShutdownTest.java | 69 +++-- 3 files changed, 272 insertions(+), 44 deletions(-) create mode 100644 framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java diff --git a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java index 97bd4e03c21..df953f34387 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java @@ -2,6 +2,7 @@ import java.util.Collections; import java.util.List; +import java.util.concurrent.ExecutorService; import java.util.function.BooleanSupplier; import javax.annotation.Resource; import org.junit.Assert; @@ -14,6 +15,7 @@ import org.tron.common.parameter.CommonParameter; import org.tron.common.runtime.vm.DataWord; import org.tron.common.runtime.vm.LogInfo; +import org.tron.common.utils.ReflectUtils; import org.tron.core.config.args.Args; import org.tron.core.services.jsonrpc.TronJsonRpc.FilterRequest; import org.tron.core.services.jsonrpc.TronJsonRpc.LogFilterElement; @@ -59,10 +61,10 @@ private static void await(BooleanSupplier condition, String message) } @Test - public void consumerThreadStartedOnce() { - long count = Thread.getAllStackTraces().keySet().stream() - .filter(t -> "filter".equals(t.getName())).count(); - Assert.assertEquals(1, count); + public void consumerStartedWithContext() { + ExecutorService filterEs = ReflectUtils.getFieldValue(tronJsonRpc, "filterEs"); + Assert.assertNotNull("consumer did not start", filterEs); + Assert.assertFalse("consumer already shut down", filterEs.isShutdown()); } @Test diff --git a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java new file mode 100644 index 00000000000..9d6413de126 --- /dev/null +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java @@ -0,0 +1,237 @@ +package org.tron.core.jsonrpc; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; +import org.junit.After; +import org.junit.Assert; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.tron.common.TestConstants; +import org.tron.common.application.TronApplicationContext; +import org.tron.common.logsfilter.capsule.BlockFilterCapsule; +import org.tron.common.logsfilter.capsule.LogsFilterCapsule; +import org.tron.common.logsfilter.queue.FilterCapsuleQueue; +import org.tron.common.parameter.CommonParameter; +import org.tron.common.utils.ReflectUtils; +import org.tron.core.config.DefaultConfig; +import org.tron.core.config.args.Args; +import org.tron.core.services.jsonrpc.TronJsonRpcImpl; + +/** + * Lifecycle of the filter consumer owned by TronJsonRpcImpl. Only the context-destruction + * case needs Spring; the rest drive a directly constructed instance. + */ +public class FilterPipelineLifecycleTest { + + private static final String FILTER_THREAD = "filter"; + private static final String LOOP_METHOD = "filterProcessLoop"; + // shutdownAndAwaitTermination() escalates to shutdownNow() after 60s; a stop well + // below that bound can only come from the stop flag + private static final long GRACEFUL_BOUND_MS = 10_000; + + @ClassRule + public static final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final FilterCapsuleQueue queue = new FilterCapsuleQueue(); + private TronJsonRpcImpl tronJsonRpc; + private TronApplicationContext context; + + private void startStandalone(boolean filterEnabled) { + CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(filterEnabled); + CommonParameter.getInstance().setJsonRpcHttpSolidityNodeEnable(false); + tronJsonRpc = new TronJsonRpcImpl(null, null, null); + ReflectUtils.setFieldValue(tronJsonRpc, "filterCapsuleQueue", queue); + ReflectUtils.invokeMethod(tronJsonRpc, "start"); + } + + @After + public void tearDown() throws IOException { + if (tronJsonRpc != null) { + tronJsonRpc.close(); + } + if (context != null) { + if (context.isActive()) { + context.close(); + } + Args.clearParam(); + } + CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(false); + CommonParameter.getInstance().setJsonRpcHttpSolidityNodeEnable(false); + } + + private ExecutorService filterEs() { + return ReflectUtils.getFieldValue(tronJsonRpc, "filterEs"); + } + + private static Optional filterThread() { + return Thread.getAllStackTraces().entrySet().stream() + .filter(e -> FILTER_THREAD.equals(e.getKey().getName()) && Arrays.stream(e.getValue()) + .anyMatch(frame -> LOOP_METHOD.equals(frame.getMethodName()))) + .map(Map.Entry::getKey).findFirst(); + } + + private static boolean consumerRunning() { + return filterThread().isPresent(); + } + + private static void await(BooleanSupplier condition, String message) + throws InterruptedException { + long deadline = System.currentTimeMillis() + 10_000; + while (System.currentTimeMillis() < deadline) { + if (condition.getAsBoolean()) { + return; + } + Thread.sleep(50); + } + Assert.fail(message); + } + + private static long timedClose(TronJsonRpcImpl rpc) throws IOException { + long t0 = System.nanoTime(); + rpc.close(); + return (System.nanoTime() - t0) / 1_000_000; + } + + @Test + public void disabledModeStartsNoConsumerAndCloseIsSafe() throws Exception { + startStandalone(false); + Assert.assertNull("consumer started while both filter APIs are disabled", filterEs()); + + tronJsonRpc.close(); + assertPoolsTerminated(); + tronJsonRpc.close(); + assertPoolsTerminated(); + Assert.assertNull(filterEs()); + } + + private void assertPoolsTerminated() { + Assert.assertTrue("logs-filter-pool not terminated", ReflectUtils. + getFieldValue(tronJsonRpc, "logsFilterPool").isTerminated()); + Assert.assertTrue("query-section pool not terminated", ReflectUtils. + getFieldValue(tronJsonRpc, "sectionExecutor").isTerminated()); + } + + /** + * Built like FullNode (circular references disallowed); context.close() is the + * production sequence: ApplicationImpl.shutdown() closes the bean explicitly, then bean + * destruction invokes the inferred close() again. Spring logs rather than propagates a + * failure of that second call, so the consumer state below - not the absence of an + * exception - is what this pins down. + */ + @Test + public void explicitCloseThenContextDestruction() throws Exception { + Args.setParam(new String[] {"--output-directory", + temporaryFolder.newFolder().toString()}, TestConstants.TEST_CONF); + CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(true); + DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory(); + beanFactory.setAllowCircularReferences(false); + context = new TronApplicationContext(beanFactory); + context.register(DefaultConfig.class); + context.refresh(); + TronJsonRpcImpl bean = context.getBean(TronJsonRpcImpl.class); + ExecutorService beanFilterEs = ReflectUtils.getFieldValue(bean, "filterEs"); + Assert.assertNotNull("consumer did not start", beanFilterEs); + + context.close(); + Assert.assertTrue("consumer survived context close", beanFilterEs.isTerminated()); + long elapsedMs = timedClose(bean); + Assert.assertTrue("close() after destruction should be harmless and quick, took " + + elapsedMs + "ms", elapsedMs < 1_000); + } + + @Test + public void emptyQueueStopsWithoutForcedCancellation() throws Exception { + startStandalone(true); + await(FilterPipelineLifecycleTest::consumerRunning, "consumer did not start"); + + long elapsedMs = timedClose(tronJsonRpc); + Assert.assertTrue("graceful stop took " + elapsedMs + "ms", elapsedMs < GRACEFUL_BOUND_MS); + Assert.assertTrue("consumer still alive after close()", filterEs().isTerminated()); + } + + /** + * Interrupt without close(): the stop flag still says running, so only the + * InterruptedException branch can end the loop. The executor worker outlives the task, + * hence the stack-frame check instead of thread liveness. + */ + @Test + public void interruptDuringPollExitsLoop() throws Exception { + startStandalone(true); + await(FilterPipelineLifecycleTest::consumerRunning, "consumer did not start"); + + filterThread().get().interrupt(); + await(() -> !consumerRunning(), "loop kept running after interrupt"); + + queue.offer(new BlockFilterCapsule("0xafter-interrupt", false)); + Thread.sleep(500); + Assert.assertEquals("capsule consumed after the loop exited", 1, queue.stream().count()); + } + + /** + * The closing thread is interrupted while the consumer is mid-capsule: close() gives up + * waiting, keeps the interrupt status and still completes; the consumer is cancelled. + */ + @Test + public void interruptedCloserReturnsAndPreservesInterruptStatus() throws Exception { + startStandalone(true); + + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + queue.offer(new LogsFilterCapsule(150L, "0xblocked", null, + Collections.emptyList(), false, false) { + @Override + public boolean isSolidified() { + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return false; + } + }); + Assert.assertTrue("consumer did not pick up the capsule", + entered.await(10, TimeUnit.SECONDS)); + + AtomicBoolean interruptKept = new AtomicBoolean(false); + AtomicReference closeError = new AtomicReference<>(); + Thread closer = new Thread(() -> { + try { + tronJsonRpc.close(); + } catch (Throwable t) { + closeError.set(t); + } finally { + interruptKept.set(Thread.currentThread().isInterrupted()); + } + }, "test-closer"); + closer.start(); + Thread.sleep(500); + Assert.assertTrue("close() finished while a capsule was in flight", closer.isAlive()); + + closer.interrupt(); + closer.join(GRACEFUL_BOUND_MS); + Assert.assertFalse("interrupted close() did not return", closer.isAlive()); + Assert.assertNull("close() failed", closeError.get()); + Assert.assertTrue("close() swallowed the interrupt status", interruptKept.get()); + + // assert before releasing the latch: a released consumer would finish on its own, + // so only a still-blocked one proves close() cancelled it + try { + Assert.assertTrue("consumer was not cancelled", + filterEs().awaitTermination(10, TimeUnit.SECONDS)); + } finally { + release.countDown(); + } + } +} diff --git a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java index b58e95a5784..74f265255a2 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java @@ -4,43 +4,44 @@ import java.io.UncheckedIOException; import java.util.Collections; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; -import javax.annotation.Resource; +import org.junit.After; import org.junit.Assert; -import org.junit.FixMethodOrder; +import org.junit.Before; import org.junit.Test; -import org.junit.runners.MethodSorters; -import org.tron.common.BaseTest; -import org.tron.common.TestConstants; 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.runtime.vm.LogInfo; -import org.tron.core.config.args.Args; +import org.tron.common.utils.ReflectUtils; import org.tron.core.services.jsonrpc.TronJsonRpc.FilterRequest; import org.tron.core.services.jsonrpc.TronJsonRpcImpl; import org.tron.core.services.jsonrpc.filters.LogFilterAndResult; import org.tron.protos.Protocol.TransactionInfo; /** - * Shutdown semantics of the decoupled filter pipeline. Method order matters: - * test1 closes the shared TronJsonRpcImpl bean, test2 asserts the second close - * is a guarded no-op, so they run in name order. + * Shutdown semantics of the decoupled filter pipeline, on a directly constructed instance. */ -@FixMethodOrder(MethodSorters.NAME_ASCENDING) -public class FilterPipelineShutdownTest extends BaseTest { +public class FilterPipelineShutdownTest { - static { - Args.setParam(new String[] {"--output-directory", dbPath()}, TestConstants.TEST_CONF); - // isJsonRpcFilterEnabled() must hold at context startup so the consumer thread starts + private final FilterCapsuleQueue filterCapsuleQueue = new FilterCapsuleQueue(); + private TronJsonRpcImpl tronJsonRpc; + + @Before + public void setUp() { CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(true); + tronJsonRpc = new TronJsonRpcImpl(null, null, null); + ReflectUtils.setFieldValue(tronJsonRpc, "filterCapsuleQueue", filterCapsuleQueue); + ReflectUtils.invokeMethod(tronJsonRpc, "start"); } - @Resource - private TronJsonRpcImpl tronJsonRpc; - @Resource - private FilterCapsuleQueue filterCapsuleQueue; + @After + public void tearDown() throws IOException { + tronJsonRpc.close(); + CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(false); + } private static TransactionInfo buildTxInfoWithLog() { LogInfo logInfo = new LogInfo(new byte[20], @@ -51,10 +52,11 @@ private static TransactionInfo buildTxInfoWithLog() { /** * close() while the consumer is mid-capsule on the parallel (logsFilterPool) path: * the in-flight capsule must complete instead of dying on a RejectedExecutionException, - * which is exactly the filterEs-before-logsFilterPool ordering constraint. + * which is exactly the filterEs-before-logsFilterPool ordering constraint. A capsule + * queued behind it is not drained. */ @Test - public void test1CloseDuringParallelProcessingLosesNoEvent() throws Exception { + public void closeDuringParallelProcessingLosesNoEvent() throws Exception { tronJsonRpc.setFilterParallelThreshold(0); LogFilterAndResult filter = new LogFilterAndResult(new FilterRequest(), 100L, null); tronJsonRpc.getEventFilter2ResultFull().put("shutdown-race", filter); @@ -78,6 +80,9 @@ public boolean isSolidified() { filterCapsuleQueue.offer(capsule); Assert.assertTrue("consumer did not pick up the capsule", entered.await(10, TimeUnit.SECONDS)); + LogsFilterCapsule queued = new LogsFilterCapsule(151L, "0xqueued", null, + Collections.singletonList(buildTxInfoWithLog()), false, false); + filterCapsuleQueue.offer(queued); Thread closer = new Thread(() -> { try { @@ -94,27 +99,11 @@ public boolean isSolidified() { release.countDown(); closer.join(150_000); Assert.assertFalse("close() did not finish", closer.isAlive()); + ExecutorService filterEs = ReflectUtils.getFieldValue(tronJsonRpc, "filterEs"); + Assert.assertTrue("consumer still running after close()", filterEs.isTerminated()); Assert.assertEquals("in-flight capsule was lost during close()", 1, filter.getResult().size()); - } - - @Test - public void test2RepeatedCloseIsIdempotentAndThreadStopped() throws Exception { - long deadline = System.currentTimeMillis() + 10_000; - while (System.currentTimeMillis() < deadline && filterThreadAlive()) { - Thread.sleep(50); - } - Assert.assertFalse("consumer thread still alive after close()", filterThreadAlive()); - - long t0 = System.nanoTime(); - tronJsonRpc.close(); - long elapsedMs = (System.nanoTime() - t0) / 1_000_000; - Assert.assertTrue("repeated close should be a fast no-op, took " + elapsedMs + "ms", - elapsedMs < 1_000); - } - - private static boolean filterThreadAlive() { - return Thread.getAllStackTraces().keySet().stream() - .anyMatch(t -> "filter".equals(t.getName()) && t.isAlive()); + Assert.assertTrue("queued capsule was drained during close()", + filterCapsuleQueue.stream().anyMatch(c -> c == queued)); } } From 9d81908a92843cf3b59fb1c587d7d404423304eb Mon Sep 17 00:00:00 2001 From: 0xbigapple Date: Tue, 22 Sep 2026 13:22:01 +0800 Subject: [PATCH 5/8] refactor(jsonrpc): narrow close() and await close state --- .../common/application/ApplicationImpl.java | 7 +---- .../services/jsonrpc/TronJsonRpcImpl.java | 3 +- .../jsonrpc/FilterPipelineLifecycleTest.java | 25 ++++++++-------- .../jsonrpc/FilterPipelineShutdownTest.java | 29 +++++++++---------- 4 files changed, 28 insertions(+), 36 deletions(-) diff --git a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java index 88c15ae26fc..f192214dc2f 100644 --- a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java +++ b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java @@ -67,12 +67,7 @@ public void shutdown() { 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); - } + tronJsonRpc.close(); dbManager.close(); shutdown.countDown(); } diff --git a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java index 890ffb9081a..b7390a4e41b 100644 --- a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java +++ b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java @@ -23,7 +23,6 @@ import com.google.protobuf.ByteString; import com.google.protobuf.GeneratedMessageV3; import java.io.Closeable; -import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -1647,7 +1646,7 @@ public Object[] getFilterResult(String filterId, Map { - try { - tronJsonRpc.close(); - } catch (IOException e) { - throw new UncheckedIOException(e); - } - }, "test-closer"); + ExecutorService filterEs = ReflectUtils.getFieldValue(tronJsonRpc, "filterEs"); + Thread closer = new Thread(tronJsonRpc::close, "test-closer"); closer.start(); - // close() must be parked awaiting the consumer, with logsFilterPool still open - Thread.sleep(500); - Assert.assertTrue("close() finished while a capsule was in flight", closer.isAlive()); + try { + // close() must be parked awaiting the consumer, with logsFilterPool still open + long deadline = System.currentTimeMillis() + 10_000; + while (!filterEs.isShutdown() && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + Assert.assertTrue("close() did not start stopping the consumer", filterEs.isShutdown()); + Assert.assertTrue("close() finished while a capsule was in flight", closer.isAlive()); + } finally { + release.countDown(); + } - release.countDown(); closer.join(150_000); Assert.assertFalse("close() did not finish", closer.isAlive()); - ExecutorService filterEs = ReflectUtils.getFieldValue(tronJsonRpc, "filterEs"); Assert.assertTrue("consumer still running after close()", filterEs.isTerminated()); Assert.assertEquals("in-flight capsule was lost during close()", 1, filter.getResult().size()); From 7174c6ec2b282240ef18a492af72a50f3fa13e7f Mon Sep 17 00:00:00 2001 From: 0xbigapple Date: Thu, 8 Oct 2026 16:58:23 +0800 Subject: [PATCH 6/8] fix(jsonrpc): stop filter consumer before logs-filter-pool on interrupted close --- .../services/jsonrpc/TronJsonRpcImpl.java | 7 ++- .../jsonrpc/FilterPipelineLifecycleTest.java | 47 +++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java index b7390a4e41b..42cb637cfc8 100644 --- a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java +++ b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java @@ -20,6 +20,7 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; +import com.google.common.util.concurrent.Uninterruptibles; import com.google.protobuf.ByteString; import com.google.protobuf.GeneratedMessageV3; import java.io.Closeable; @@ -1651,8 +1652,12 @@ public void close() { return; } // The consumer loop submits to logsFilterPool (over-threshold path), so it must - // terminate before the pool shuts down. + // 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); + } ExecutorServiceManager.shutdownAndAwaitTermination(logsFilterPool, "logs-filter-pool"); logElementCache.invalidateAll(); blockHashCache.invalidateAll(); diff --git a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java index 7b73a8eea44..d6e23c7f700 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java @@ -1,5 +1,6 @@ package org.tron.core.jsonrpc; +import com.google.common.util.concurrent.Uninterruptibles; import java.util.Arrays; import java.util.Collections; import java.util.Map; @@ -15,6 +16,7 @@ import org.junit.ClassRule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import org.mockito.Mockito; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.tron.common.TestConstants; import org.tron.common.application.TronApplicationContext; @@ -26,6 +28,7 @@ import org.tron.core.config.DefaultConfig; import org.tron.core.config.args.Args; import org.tron.core.services.jsonrpc.TronJsonRpcImpl; +import org.tron.core.services.jsonrpc.filters.LogFilterAndResult; /** * Lifecycle of the filter consumer owned by TronJsonRpcImpl. Only the context-destruction @@ -233,4 +236,48 @@ public boolean isSolidified() { release.countDown(); } } + + /** + * Interrupted closer with an over-threshold filter map: the in-flight capsule is handed + * to logsFilterPool, so the pool must outlive the consumer. + */ + @Test + public void interruptedCloserKeepsPoolUntilConsumerExits() throws Exception { + startStandalone(true); + tronJsonRpc.setFilterParallelThreshold(1); + LogFilterAndResult expired = Mockito.mock(LogFilterAndResult.class); + Mockito.when(expired.isExpire()).thenReturn(true); + Map filters = tronJsonRpc.getEventFilter2ResultFull(); + filters.put("0x1", expired); + filters.put("0x2", expired); + + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + queue.offer(new LogsFilterCapsule(150L, "0xblocked", null, + Collections.emptyList(), false, false) { + @Override + public boolean isSolidified() { + entered.countDown(); + Uninterruptibles.awaitUninterruptibly(release); + return false; + } + }); + Assert.assertTrue("consumer did not pick up the capsule", + entered.await(10, TimeUnit.SECONDS)); + + Thread closer = new Thread(tronJsonRpc::close, "test-closer"); + closer.start(); + try { + 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(); + } + 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()); + } } From 9bbb89f8e4b1bf6c127d91d39c55f5a53aaedf49 Mon Sep 17 00:00:00 2001 From: 0xbigapple Date: Thu, 8 Oct 2026 17:28:42 +0800 Subject: [PATCH 7/8] test(jsonrpc): scope filter consumer lookup to the instance under test --- .../core/jsonrpc/FilterPipelineLifecycleTest.java | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java index d6e23c7f700..3f7407db156 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java @@ -5,6 +5,7 @@ import java.util.Collections; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; @@ -47,9 +48,11 @@ public class FilterPipelineLifecycleTest { private final FilterCapsuleQueue queue = new FilterCapsuleQueue(); private TronJsonRpcImpl tronJsonRpc; + private Set threadsBeforeStart; private TronApplicationContext context; private void startStandalone(boolean filterEnabled) { + threadsBeforeStart = Thread.getAllStackTraces().keySet(); CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(filterEnabled); CommonParameter.getInstance().setJsonRpcHttpSolidityNodeEnable(false); tronJsonRpc = new TronJsonRpcImpl(null, null, null); @@ -76,14 +79,16 @@ private ExecutorService filterEs() { return ReflectUtils.getFieldValue(tronJsonRpc, "filterEs"); } - private static Optional filterThread() { + // only a thread started by this instance, not a consumer leaked by another test + private Optional filterThread() { return Thread.getAllStackTraces().entrySet().stream() - .filter(e -> FILTER_THREAD.equals(e.getKey().getName()) && Arrays.stream(e.getValue()) + .filter(e -> !threadsBeforeStart.contains(e.getKey()) + && FILTER_THREAD.equals(e.getKey().getName()) && Arrays.stream(e.getValue()) .anyMatch(frame -> LOOP_METHOD.equals(frame.getMethodName()))) .map(Map.Entry::getKey).findFirst(); } - private static boolean consumerRunning() { + private boolean consumerRunning() { return filterThread().isPresent(); } @@ -155,7 +160,7 @@ public void explicitCloseThenContextDestruction() throws Exception { @Test public void emptyQueueStopsWithoutForcedCancellation() throws Exception { startStandalone(true); - await(FilterPipelineLifecycleTest::consumerRunning, "consumer did not start"); + await(this::consumerRunning, "consumer did not start"); long elapsedMs = timedClose(tronJsonRpc); Assert.assertTrue("graceful stop took " + elapsedMs + "ms", elapsedMs < GRACEFUL_BOUND_MS); @@ -170,7 +175,7 @@ public void emptyQueueStopsWithoutForcedCancellation() throws Exception { @Test public void interruptDuringPollExitsLoop() throws Exception { startStandalone(true); - await(FilterPipelineLifecycleTest::consumerRunning, "consumer did not start"); + await(this::consumerRunning, "consumer did not start"); filterThread().get().interrupt(); await(() -> !consumerRunning(), "loop kept running after interrupt"); From 3290dcb25256665d432d2c99d33669566ba62b89 Mon Sep 17 00:00:00 2001 From: 0xbigapple Date: Thu, 8 Oct 2026 19:30:45 +0800 Subject: [PATCH 8/8] docs(framework): state the json-rpc filter consumer shutdown ordering --- .../main/java/org/tron/common/application/ApplicationImpl.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java index f192214dc2f..a7c0fd5ca87 100644 --- a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java +++ b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java @@ -66,7 +66,8 @@ public void shutdown() { if (solidityNode != null) { solidityNode.close(); } - // producers are stopped; stop the json-rpc filter consumer before the DB closes + // Stop the json-rpc filter consumer explicitly, once the producers above are stopped, + // rather than in Spring's destroy phase. tronJsonRpc.close(); dbManager.close(); shutdown.countDown();