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..a7c0fd5ca87 100644 --- a/framework/src/main/java/org/tron/common/application/ApplicationImpl.java +++ b/framework/src/main/java/org/tron/common/application/ApplicationImpl.java @@ -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") @@ -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); /** @@ -62,6 +66,9 @@ public void shutdown() { if (solidityNode != null) { solidityNode.close(); } + // 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(); } 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..35f551966f1 --- /dev/null +++ b/framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java @@ -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 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); + } + + @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 9d7a7c979b9..d4dbddf2768 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 @@ -2344,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, @@ -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); } } @@ -2658,7 +2614,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..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,10 +20,10 @@ 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; -import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -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,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(); + return; + } catch (Throwable throwable) { + logger.error("Unknown throwable happened in filterProcessLoop. ", throwable); + } + } } @VisibleForTesting @@ -1613,7 +1647,17 @@ public Object[] getFilterResult(String filterId, Map 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, @@ -1868,7 +1866,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 +1876,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..df953f34387 --- /dev/null +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java @@ -0,0 +1,102 @@ +package org.tron.core.jsonrpc; + +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; +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.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; +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 consumerStartedWithContext() { + ExecutorService filterEs = ReflectUtils.getFieldValue(tronJsonRpc, "filterEs"); + Assert.assertNotNull("consumer did not start", filterEs); + Assert.assertFalse("consumer already shut down", filterEs.isShutdown()); + } + + @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/FilterPipelineLifecycleTest.java b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java new file mode 100644 index 00000000000..3f7407db156 --- /dev/null +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineLifecycleTest.java @@ -0,0 +1,288 @@ +package org.tron.core.jsonrpc; + +import com.google.common.util.concurrent.Uninterruptibles; +import java.util.Arrays; +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; +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.mockito.Mockito; +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; +import org.tron.core.services.jsonrpc.filters.LogFilterAndResult; + +/** + * 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 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); + ReflectUtils.setFieldValue(tronJsonRpc, "filterCapsuleQueue", queue); + ReflectUtils.invokeMethod(tronJsonRpc, "start"); + } + + @After + public void tearDown() { + 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"); + } + + // only a thread started by this instance, not a consumer leaked by another test + private Optional filterThread() { + return Thread.getAllStackTraces().entrySet().stream() + .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 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) { + 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(this::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(this::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(); + try { + await(filterEs()::isShutdown, "close() did not start stopping the consumer"); + 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 + Assert.assertTrue("consumer was not cancelled", + filterEs().awaitTermination(10, TimeUnit.SECONDS)); + } finally { + 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()); + } +} 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..3630653e717 --- /dev/null +++ b/framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineShutdownTest.java @@ -0,0 +1,108 @@ +package org.tron.core.jsonrpc; + +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +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.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, on a directly constructed instance. + */ +public class FilterPipelineShutdownTest { + + 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"); + } + + @After + public void tearDown() { + tronJsonRpc.close(); + CommonParameter.getInstance().setJsonRpcHttpFullNodeEnable(false); + } + + 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. A capsule + * queued behind it is not drained. + */ + @Test + public void closeDuringParallelProcessingLosesNoEvent() 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)); + LogsFilterCapsule queued = new LogsFilterCapsule(151L, "0xqueued", null, + Collections.singletonList(buildTxInfoWithLog()), false, false); + filterCapsuleQueue.offer(queued); + + ExecutorService filterEs = ReflectUtils.getFieldValue(tronJsonRpc, "filterEs"); + Thread closer = new Thread(tronJsonRpc::close, "test-closer"); + closer.start(); + 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(); + } + + closer.join(150_000); + Assert.assertFalse("close() did not finish", closer.isAlive()); + Assert.assertTrue("consumer still running after close()", filterEs.isTerminated()); + Assert.assertEquals("in-flight capsule was lost during close()", + 1, filter.getResult().size()); + Assert.assertTrue("queued capsule was drained during close()", + filterCapsuleQueue.stream().anyMatch(c -> c == queued)); + } +} 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<>();