diff --git a/METRICS_CHANGELOG.md b/METRICS_CHANGELOG.md index 3c599796d7a..b897f6c5d26 100644 --- a/METRICS_CHANGELOG.md +++ b/METRICS_CHANGELOG.md @@ -3,6 +3,17 @@ Metrics Changelog This file tracks Prometheus metric additions, changes, and removals in java-tron. For the full set of metrics emitted today, see the references at the bottom. +**4.8.3** + +### New Metrics + +#### Core + +- `tron:node_info` (Info, labels `version`, `genesis_block_id`) — static node identity: the running node version string plus the full genesis block hash as the canonical chain identifier, for fleet dashboards and version/chain alert rules. ([#6923](https://github.com/tronprotocol/java-tron/issues/6923)) +- `tron:block_fetch_armed` (Counter) — incremented when `FetchBlockService` arms a fetch tracking for a block; denominator for secondary-fetch rates. ([#6923](https://github.com/tronprotocol/java-tron/issues/6923)) +- `tron:block_fetch_secondary` (Counter) — incremented when a secondary fetch request is sent to an alternate peer. ([#6923](https://github.com/tronprotocol/java-tron/issues/6923)) +- `tron:block_already_known` (Counter) — incremented when a block response matches an outstanding adv request whose exact block ID is already known before that response is processed (best-effort signal; concurrent arrivals may be missed). ([#6923](https://github.com/tronprotocol/java-tron/issues/6923)) + **4.8.2** ### New Metrics diff --git a/common/src/main/java/org/tron/common/prometheus/MetricKeys.java b/common/src/main/java/org/tron/common/prometheus/MetricKeys.java index 95a38c4b479..d392a47d200 100644 --- a/common/src/main/java/org/tron/common/prometheus/MetricKeys.java +++ b/common/src/main/java/org/tron/common/prometheus/MetricKeys.java @@ -21,6 +21,10 @@ public static class Counter { public static final String P2P_ERROR = "tron:p2p_error"; public static final String P2P_DISCONNECT = "tron:p2p_disconnect"; public static final String INTERNAL_SERVICE_FAIL = "tron:internal_service_fail"; + // verification counters for the bounded fetch latency estimator rollout + public static final String BLOCK_FETCH_ARMED = "tron:block_fetch_armed"; + public static final String BLOCK_FETCH_SECONDARY = "tron:block_fetch_secondary"; + public static final String BLOCK_ALREADY_KNOWN = "tron:block_already_known"; private Counter() { throw new IllegalStateException("Counter"); @@ -44,6 +48,16 @@ private Gauge() { } + // Info + public static class Info { + public static final String NODE_INFO = "tron:node"; + + private Info() { + throw new IllegalStateException("Info"); + } + + } + // Histogram public static class Histogram { public static final String HTTP_SERVICE_LATENCY = "tron:http_service_latency_seconds"; diff --git a/common/src/main/java/org/tron/common/prometheus/MetricLabels.java b/common/src/main/java/org/tron/common/prometheus/MetricLabels.java index 1f0da214085..7f3d5fa6076 100644 --- a/common/src/main/java/org/tron/common/prometheus/MetricLabels.java +++ b/common/src/main/java/org/tron/common/prometheus/MetricLabels.java @@ -78,4 +78,17 @@ private Histogram() { } + // Info + public static class Info { + public static final String VERSION = "version"; + // identifies the genesis block: the label value is the chain id derived from the + // genesis block hash, so the label is named genesis_block_id + public static final String CHAIN_ID = "genesis_block_id"; + + private Info() { + throw new IllegalStateException("Info"); + } + + } + } diff --git a/common/src/main/java/org/tron/common/prometheus/Metrics.java b/common/src/main/java/org/tron/common/prometheus/Metrics.java index 6774dd7c315..d3506684231 100644 --- a/common/src/main/java/org/tron/common/prometheus/Metrics.java +++ b/common/src/main/java/org/tron/common/prometheus/Metrics.java @@ -65,4 +65,8 @@ public static void histogramObserve(Histogram.Timer startTimer) { public static void histogramObserve(String key, double amt, String... labels) { MetricsHistogram.observe(key, amt, labels); } + + public static void info(String key, String... labels) { + MetricsInfo.set(key, labels); + } } diff --git a/common/src/main/java/org/tron/common/prometheus/MetricsCounter.java b/common/src/main/java/org/tron/common/prometheus/MetricsCounter.java index 7231baaba8f..2e066064ed8 100644 --- a/common/src/main/java/org/tron/common/prometheus/MetricsCounter.java +++ b/common/src/main/java/org/tron/common/prometheus/MetricsCounter.java @@ -19,6 +19,15 @@ class MetricsCounter { init(MetricKeys.Counter.P2P_DISCONNECT, "tron p2p disconnect .", "type"); init(MetricKeys.Counter.INTERNAL_SERVICE_FAIL, "internal Service fail.", "class", "method"); + init(MetricKeys.Counter.BLOCK_FETCH_ARMED, + "in-flight fetch requests armed by the fetch-block service."); + init(MetricKeys.Counter.BLOCK_FETCH_SECONDARY, + "secondary fetch requests issued by the fetch-block failover estimator."); + init(MetricKeys.Counter.BLOCK_ALREADY_KNOWN, + "adv block responses matched to an outstanding request whose exact block id " + + "was already known when the response was handled (best-effort: concurrent " + + "arrivals may be missed; a duplicate is not attributed to secondary " + + "fetches)."); } private MetricsCounter() { diff --git a/common/src/main/java/org/tron/common/prometheus/MetricsInfo.java b/common/src/main/java/org/tron/common/prometheus/MetricsInfo.java new file mode 100644 index 00000000000..e3cd8b7ca65 --- /dev/null +++ b/common/src/main/java/org/tron/common/prometheus/MetricsInfo.java @@ -0,0 +1,40 @@ +package org.tron.common.prometheus; + +import io.prometheus.client.Info; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import lombok.extern.slf4j.Slf4j; + +@Slf4j(topic = "metrics") +class MetricsInfo { + + private static final Map container = new ConcurrentHashMap<>(); + + static { + init(MetricKeys.Info.NODE_INFO, "tron node info.", + MetricLabels.Info.VERSION, MetricLabels.Info.CHAIN_ID); + } + + private MetricsInfo() { + throw new IllegalStateException("MetricsInfo"); + } + + private static void init(String name, String help, String... labels) { + container.put(name, Info.build() + .name(name) + .help(help) + .labelNames(labels) + .register()); + } + + static void set(String key, String... labels) { + if (Metrics.enabled()) { + Info info = container.get(key); + if (info == null) { + logger.info("{} not exist", key); + return; + } + info.labels(labels); + } + } +} diff --git a/framework/src/main/java/org/tron/core/metrics/MetricsKey.java b/framework/src/main/java/org/tron/core/metrics/MetricsKey.java index 3ac7b5840d8..c4630994b3f 100644 --- a/framework/src/main/java/org/tron/core/metrics/MetricsKey.java +++ b/framework/src/main/java/org/tron/core/metrics/MetricsKey.java @@ -23,6 +23,5 @@ public class MetricsKey { public static final String NET_API_DETAIL_QPS = "net.api.detail.qps."; public static final String NET_API_DETAIL_FAIL_QPS = "net.api.detail.failQps."; public static final String NET_API_DETAIL_OUT_TRAFFIC = "net.api.detail.outTraffic."; - public static final String NET_LATENCY_FETCH_BLOCK = "net.latency.fetch.block."; } diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java index 452209d575f..b1631979913 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java @@ -15,8 +15,6 @@ import org.tron.core.config.args.Args; import org.tron.core.exception.P2pException; import org.tron.core.exception.P2pException.TypeEnum; -import org.tron.core.metrics.MetricsKey; -import org.tron.core.metrics.MetricsUtil; import org.tron.core.net.TronNetDelegate; import org.tron.core.net.message.TronMessage; import org.tron.core.net.message.adv.BlockMessage; @@ -91,10 +89,19 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep } Long time = peer.getAdvInvRequest().remove(item); if (null != time) { - MetricsUtil.histogramUpdateUnCheck(MetricsKey.NET_LATENCY_FETCH_BLOCK - + peer.getInetAddress(), now - time); + peer.updateFetchLatency(now - time); Metrics.histogramObserve(MetricKeys.Histogram.BLOCK_FETCH_LATENCY, (now - time) / Metrics.MILLISECONDS_PER_SECOND); + // Best-effort duplicate signal: only responses matched to an outstanding adv + // request whose exact block id was already known before this response is + // processed (a concurrent or redundant arrival) are counted. The lookup is + // exact-id (block store + khaos), not a height comparison: an unknown fork + // block below head must not count. Concurrency can still let a simultaneous + // arrival slip through, and a duplicate is not attributed to a secondary + // fetch, so this is a lower-bound indicator rather than an exact count. + if (tronNetDelegate.containBlock(blockId)) { + Metrics.counterInc(MetricKeys.Counter.BLOCK_ALREADY_KNOWN, 1); + } } Metrics.histogramObserve(MetricKeys.Histogram.BLOCK_RECEIVE_DELAY, (now - blockMessage.getBlockCapsule().getTimeStamp()) / Metrics.MILLISECONDS_PER_SECOND); diff --git a/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java b/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java index 7d7457cf2fc..dedf1c23f8e 100644 --- a/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java +++ b/framework/src/main/java/org/tron/core/net/peer/PeerConnection.java @@ -24,7 +24,9 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Scope; import org.springframework.stereotype.Component; +import org.tron.common.math.StrictMathWrapper; import org.tron.common.overlay.message.Message; +import org.tron.common.parameter.CommonParameter; import org.tron.common.prometheus.MetricKeys; import org.tron.common.prometheus.Metrics; import org.tron.common.utils.Pair; @@ -92,6 +94,21 @@ public class PeerConnection { @Getter private volatile long blockRcvTime; + /** + * EWMA smoothing divisor for the fetch latency estimator: the previous estimate is + * weighted (EWMA_DIVISOR - 1) / EWMA_DIVISOR and the new sample 1 / EWMA_DIVISOR, + * i.e. alpha = 0.1. This trades off smoothing against responsiveness and sits in the + * same order of magnitude as TCP's SRTT gain (1/8, RFC 6298). Under a large + * degradation the relative ordering of two peers can flip within 1-2 samples, while + * the absolute value converges smoothly (e.g. seeded at 100, ten 500ms samples walk + * 140, 176, 208, 237, 263, 286, 307, 326, 343, 358 without ever hitting the clamp). + */ + private static final int EWMA_DIVISOR = 10; + + private volatile long fetchLatency; + + private volatile boolean fetchLatencySeeded; + @Getter @Setter private volatile TronState tronState = TronState.INIT; @@ -184,6 +201,49 @@ public void setChannel(Channel channel) { Args.getInstance().getRateLimiterDisconnect()); } + /** + * Bounded fetch latency estimator with an explicit unsampled state. + * + *

The channel's average latency is never part of the sample sequence; it is only a + * read fallback while the estimator is unsampled (see {@link #getFetchLatency()}). The + * first measured fetch latency directly replaces the unsampled state (isomorphic to + * RFC 6298 SRTT initialization), and subsequent samples are blended with an EWMA of + * alpha = 1 / EWMA_DIVISOR = 0.1. With integer division the EWMA has a fixed point, e.g. + * (499 * 9 + 500) / 10 = 499, which damps jitter around the saturation bound. + * + *

A single fetch worker reads this value while the channel event loop writes it; + * volatile is sufficient for this benign race and no lock should be added. + * + * @param latencyMillis measured fetch latency in milliseconds + */ + public void updateFetchLatency(long latencyMillis) { + if (!fetchLatencySeeded) { + fetchLatency = clampFetchLatency(latencyMillis); + fetchLatencySeeded = true; + } else { + fetchLatency = clampFetchLatency( + (fetchLatency * (EWMA_DIVISOR - 1) + latencyMillis) / EWMA_DIVISOR); + } + } + + /** + * Returns the bounded fetch latency estimate. While the estimator has not observed a + * real fetch sample yet, the channel's average latency is returned as a read fallback + * (an unknown peer is treated via its transport-level estimate instead of 0). + */ + public long getFetchLatency() { + if (!fetchLatencySeeded) { + return channel.getAvgLatency(); + } + return fetchLatency; + } + + private long clampFetchLatency(long latency) { + // Saturation intentionally makes the >= timeout gate in FetchBlockService trigger. + return StrictMathWrapper.max(0, + StrictMathWrapper.min(CommonParameter.getInstance().fetchBlockTimeout, latency)); + } + public void setBlockBothHave(BlockId blockId) { this.blockBothHave = blockId; this.blockBothHaveUpdateTime = System.currentTimeMillis(); diff --git a/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java b/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java index bda2646abbc..4642cd0bdf4 100644 --- a/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java +++ b/framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java @@ -13,11 +13,11 @@ import org.springframework.stereotype.Component; import org.tron.common.es.ExecutorServiceManager; import org.tron.common.parameter.CommonParameter; +import org.tron.common.prometheus.MetricKeys; +import org.tron.common.prometheus.Metrics; import org.tron.common.utils.Sha256Hash; import org.tron.core.ChainBaseManager; import org.tron.core.capsule.BlockCapsule; -import org.tron.core.metrics.MetricsKey; -import org.tron.core.metrics.MetricsUtil; import org.tron.core.net.TronNetDelegate; import org.tron.core.net.message.adv.FetchInvDataMessage; import org.tron.core.net.peer.Item; @@ -73,6 +73,7 @@ public void fetchBlock(List sha256HashList, PeerConnection peer) { .findFirst().ifPresent(sha256Hash -> { long now = System.currentTimeMillis(); fetchBlockInfo = new FetchBlockInfo(sha256Hash, peer, now); + Metrics.counterInc(MetricKeys.Counter.BLOCK_FETCH_ARMED, 1); logger.info("Set fetchBlockInfo, block: {}, peer: {}, time: {}", sha256Hash, peer.getInetAddress(), now); }); @@ -97,9 +98,9 @@ private void fetchBlockProcess(FetchBlockInfo fetchBlock) { .filter(PeerConnection::isIdle) .filter(filterPeer -> !filterPeer.equals(fetchBlock.getPeer())) .filter(filterPeer -> filterPeer.getAdvInvReceive().getIfPresent(item) != null) - .filter(filterPeer -> getPeerTop75(filterPeer) - <= CommonParameter.getInstance().fetchBlockTimeout) - .min(Comparator.comparingDouble(this::getPeerTop75)); + // Seeded estimates are clamped to the fetch timeout; the unseeded channel-latency + // fallback is not, but min() ordering and the saturation gate keep it safe. + .min(Comparator.comparingDouble(this::getPeerLatency)); if (optionalPeerConnection.isPresent()) { optionalPeerConnection.ifPresent(firstPeer -> { @@ -107,6 +108,7 @@ private void fetchBlockProcess(FetchBlockInfo fetchBlock) { && firstPeer.checkAndPutAdvInvRequest(item, System.currentTimeMillis())) { firstPeer.sendMessage(new FetchInvDataMessage(Collections.singletonList(item.getHash()), item.getType())); + Metrics.counterInc(MetricKeys.Counter.BLOCK_FETCH_SECONDARY, 1); this.fetchBlockInfo = null; } }); @@ -120,21 +122,27 @@ private void fetchBlockProcess(FetchBlockInfo fetchBlock) { } private boolean shouldFetchBlock(PeerConnection newPeer, FetchBlockInfo fetchBlock) { - double newPeerTop75 = getPeerTop75(newPeer); - double oldPeerTop75 = getPeerTop75(fetchBlock.getPeer()); + double newPeerLatency = getPeerLatency(newPeer); + double oldPeerLatency = getPeerLatency(fetchBlock.getPeer()); long oldPeerSpendTime = System.currentTimeMillis() - fetchBlock.getTime(); - if (oldPeerTop75 > fetchTimeOut || oldPeerSpendTime >= fetchTimeOut) { + // Switch unconditionally on a hard timeout: an unseeded or saturated old peer must not + // permanently wedge fetchBlockInfo. + if (oldPeerSpendTime >= fetchTimeOut) { return true; } - double oldPeerLeftTime = oldPeerTop75 - oldPeerSpendTime; - return newPeerTop75 < oldPeerLeftTime * BLOCK_FETCH_LEFT_TIME_PERCENT - && oldPeerSpendTime + newPeerTop75 < fetchTimeOut; + // Require a strictly better peer for the latency saturation gate to prevent 500v500 flapping. + if (oldPeerLatency >= fetchTimeOut && newPeerLatency < oldPeerLatency) { + return true; + } + + double oldPeerLeftTime = oldPeerLatency - oldPeerSpendTime; + return newPeerLatency < oldPeerLeftTime * BLOCK_FETCH_LEFT_TIME_PERCENT + && oldPeerSpendTime + newPeerLatency < fetchTimeOut; } - private double getPeerTop75(PeerConnection peerConnection) { - return MetricsUtil.getHistogram(MetricsKey.NET_LATENCY_FETCH_BLOCK - + peerConnection.getInetAddress()).getSnapshot().get75thPercentile(); + private double getPeerLatency(PeerConnection peerConnection) { + return peerConnection.getFetchLatency(); } private static class FetchBlockInfo { @@ -159,4 +167,4 @@ public FetchBlockInfo(Sha256Hash hash, PeerConnection peer, long time) { } -} \ No newline at end of file +} diff --git a/framework/src/main/java/org/tron/core/services/RpcApiService.java b/framework/src/main/java/org/tron/core/services/RpcApiService.java index b9cb05a3b14..4949f62a7ab 100755 --- a/framework/src/main/java/org/tron/core/services/RpcApiService.java +++ b/framework/src/main/java/org/tron/core/services/RpcApiService.java @@ -10,6 +10,7 @@ import io.grpc.netty.NettyServerBuilder; import io.grpc.stub.StreamObserver; import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -2655,9 +2656,15 @@ public void getBlock(GrpcAPI.BlockReq request, public class MonitorApi extends MonitorGrpc.MonitorImplBase { + private final AtomicBoolean deprecatedWarned = new AtomicBoolean(false); + @Override public void getStatsInfo(EmptyMessage request, StreamObserver responseObserver) { + if (deprecatedWarned.compareAndSet(false, true)) { + logger.warn("rpc Monitor.GetStatsInfo is deprecated and will be removed in a " + + "future major release; migrate to the prometheus metrics endpoint"); + } responseObserver.onNext(metricsApiService.getMetricProtoInfo()); responseObserver.onCompleted(); } diff --git a/framework/src/main/java/org/tron/core/services/http/MetricsServlet.java b/framework/src/main/java/org/tron/core/services/http/MetricsServlet.java index aaaebb22146..0d14b3abc49 100644 --- a/framework/src/main/java/org/tron/core/services/http/MetricsServlet.java +++ b/framework/src/main/java/org/tron/core/services/http/MetricsServlet.java @@ -1,5 +1,6 @@ package org.tron.core.services.http; +import java.util.concurrent.atomic.AtomicBoolean; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; import lombok.extern.slf4j.Slf4j; @@ -13,10 +14,16 @@ @Slf4j(topic = "API") public class MetricsServlet extends RateLimiterServlet { + private static final AtomicBoolean deprecatedWarned = new AtomicBoolean(false); + @Autowired private MetricsApiService metricsApiService; protected void doGet(HttpServletRequest request, HttpServletResponse response) { + if (deprecatedWarned.compareAndSet(false, true)) { + logger.warn("HTTP /monitor/getstatsinfo is deprecated and will be removed in a " + + "future major release; migrate to the prometheus metrics endpoint"); + } try { MetricsInfo metricsInfo = metricsApiService.getMetricsInfo(); diff --git a/framework/src/main/java/org/tron/program/FullNode.java b/framework/src/main/java/org/tron/program/FullNode.java index 96b9f73d577..b7ff9f063d9 100644 --- a/framework/src/main/java/org/tron/program/FullNode.java +++ b/framework/src/main/java/org/tron/program/FullNode.java @@ -10,6 +10,7 @@ import org.tron.common.exit.ExitManager; import org.tron.common.log.LogService; import org.tron.common.parameter.CommonParameter; +import org.tron.common.prometheus.MetricKeys; import org.tron.common.prometheus.Metrics; import org.tron.core.config.DefaultConfig; import org.tron.core.config.args.Args; @@ -51,6 +52,11 @@ public static void main(String[] args) { // init metrics first Metrics.init(); + if (parameter.isNodeMetricsEnable()) { + logger.warn("legacy metrics stack (node.metricsEnable) is deprecated and will be " + + "removed in a future major release; migrate to node.metrics.prometheus.enable"); + } + DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory(); beanFactory.setAllowCircularReferences(false); TronApplicationContext context = @@ -60,6 +66,10 @@ public static void main(String[] args) { Application appT = ApplicationFactory.create(context); context.registerShutdownHook(); appT.startup(); + // the genesis block id (chainId) is only available after the context refresh + // (Manager.initGenesis) + Metrics.info(MetricKeys.Info.NODE_INFO, Version.getVersion(), + Args.getInstance().getChainId()); if (parameter.isSolidityNode()) { SolidityNode node = context.getBean(SolidityNode.class); node.run(); diff --git a/framework/src/test/java/org/tron/core/metrics/prometheus/PrometheusApiServiceTest.java b/framework/src/test/java/org/tron/core/metrics/prometheus/PrometheusApiServiceTest.java index dd260a1b869..fba687fcdcc 100644 --- a/framework/src/test/java/org/tron/core/metrics/prometheus/PrometheusApiServiceTest.java +++ b/framework/src/test/java/org/tron/core/metrics/prometheus/PrometheusApiServiceTest.java @@ -21,6 +21,7 @@ import org.tron.common.TestConstants; import org.tron.common.crypto.ECKey; import org.tron.common.parameter.CommonParameter; +import org.tron.common.prometheus.MetricKeys; import org.tron.common.prometheus.MetricLabels; import org.tron.common.prometheus.Metrics; import org.tron.common.utils.ByteArray; @@ -36,6 +37,7 @@ import org.tron.core.config.args.Args; import org.tron.core.consensus.ConsensusService; import org.tron.core.net.TronNetDelegate; +import org.tron.program.Version; import org.tron.protos.Protocol; @Slf4j(topic = "metric") @@ -206,4 +208,30 @@ private BlockCapsule createTestBlockCapsule(long time, return blockCapsule; } + @Test + public void testNodeInfoMetric() { + String version = Version.getVersion(); + String testGenesisBlockId = + "00000000000000001ebf88508a03865c71d452e25f4d51194196a1d22b6653dc"; + Metrics.info(MetricKeys.Info.NODE_INFO, version, testGenesisBlockId); + // Prometheus Info collector appends "_info" to the sample name + Double value = CollectorRegistry.defaultRegistry.getSampleValue( + "tron:node_info", + new String[] {MetricLabels.Info.VERSION, MetricLabels.Info.CHAIN_ID}, + new String[] {version, testGenesisBlockId}); + Assert.assertNotNull("tron:node_info sample should exist", value); + Assert.assertEquals(1.0, value, 0.0); + } + + @Test + public void testNodeInfoUnknownKey() { + // unknown key exercises the null-guard branch in MetricsInfo.set + Metrics.info("tron:unknown_info", "x"); + Double value = CollectorRegistry.defaultRegistry.getSampleValue( + "tron:unknown_info_info", + new String[] {"version"}, + new String[] {"x"}); + Assert.assertNull(value); + } + } \ No newline at end of file diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java new file mode 100644 index 00000000000..64e77d77bba --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java @@ -0,0 +1,179 @@ +package org.tron.core.net.messagehandler; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.prometheus.client.CollectorRegistry; +import java.net.InetSocketAddress; +import java.util.Collections; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.tron.common.parameter.CommonParameter; +import org.tron.common.prometheus.MetricKeys; +import org.tron.common.utils.ReflectUtils; +import org.tron.common.utils.Sha256Hash; +import org.tron.core.capsule.BlockCapsule; +import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.exception.P2pException; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.adv.BlockMessage; +import org.tron.core.net.peer.Item; +import org.tron.core.net.peer.PeerConnection; +import org.tron.core.net.service.adv.AdvService; +import org.tron.core.net.service.fetchblock.FetchBlockService; +import org.tron.core.net.service.relay.RelayService; +import org.tron.core.net.service.sync.SyncService; +import org.tron.core.services.WitnessProductBlockService; +import org.tron.p2p.connection.Channel; +import org.tron.protos.Protocol.Inventory.InventoryType; + +/** + * Focused unit tests for the tron:block_already_known counter semantics. + * + *

The counter only counts adv block responses that were matched to an outstanding + * adv request (the request entry is consumed exactly once) AND whose exact block id + * was already known before the response was processed. A height comparison is never + * used, because an unknown fork block can sit below head and must not count. + * + *

Pure unit tests (no Spring context): every BlockMsgHandler dependency is mocked + * and the counter value is read back from the default Prometheus registry as a delta, + * so leftover increments from earlier tests in the same JVM cannot interfere. + */ +public class BlockAlreadyKnownCounterTest { + + private static final long HEAD_NUM = 100L; + private static final String SAMPLE_NAME = MetricKeys.Counter.BLOCK_ALREADY_KNOWN + "_total"; + + private BlockMsgHandler handler; + private TronNetDelegate delegate; + private PeerConnection peer; + private boolean metricsEnabledBefore; + + @Before + public void setUp() { + metricsEnabledBefore = CommonParameter.getInstance().isMetricsPrometheusEnable(); + CommonParameter.getInstance().setMetricsPrometheusEnable(true); + delegate = mock(TronNetDelegate.class); + handler = new BlockMsgHandler(); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(handler, "advService", mock(AdvService.class)); + ReflectUtils.setFieldValue(handler, "relayService", mock(RelayService.class)); + ReflectUtils.setFieldValue(handler, "syncService", mock(SyncService.class)); + ReflectUtils.setFieldValue(handler, "fetchBlockService", mock(FetchBlockService.class)); + ReflectUtils.setFieldValue(handler, "witnessProductBlockService", + mock(WitnessProductBlockService.class)); + // production default; pinned so a leaked fast-forward flag from another test + // class in the same JVM cannot skip the no-request validation below + ReflectUtils.setFieldValue(handler, "fastForward", false); + + peer = new PeerConnection(); + Channel channel = mock(Channel.class); + InetSocketAddress address = new InetSocketAddress("127.0.0.1", 18888); + when(channel.getInetSocketAddress()).thenReturn(address); + when(channel.getInetAddress()).thenReturn(address.getAddress()); + ReflectUtils.setFieldValue(peer, "channel", channel); + } + + @After + public void tearDown() { + CommonParameter.getInstance().setMetricsPrometheusEnable(metricsEnabledBefore); + } + + @Test + public void testMatchedRequestAlreadyKnownBlockIncrements() throws P2pException { + when(delegate.containBlock(any(BlockId.class))).thenReturn(true); + when(delegate.validBlock(any(BlockCapsule.class))).thenReturn(true); + when(delegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, HEAD_NUM)); + + BlockMessage msg = newBlockMessage(1); + double before = sample(); + request(msg); + handler.processMessage(peer, msg); + assertEquals(before + 1, sample(), 0.0); + + // each matched delivery of an already-known id counts exactly once + double beforeSecond = sample(); + request(msg); + handler.processMessage(peer, msg); + assertEquals(beforeSecond + 1, sample(), 0.0); + } + + @Test + public void testMatchedRequestUnknownForkBelowHeadNotIncremented() throws P2pException { + // An unknown fork below head must not count. Only the block's own id is stubbed + // "unknown" while its parent is "known", so processing passes the unlink guard and + // reaches the low-height branch (num < head) without ever delegating: under the old + // height-based implementation that branch counted, so this test fails on it. + BlockMessage msg = newBlockMessage(1); + assertNotEquals(msg.getBlockId(), msg.getBlockCapsule().getParentBlockId()); + when(delegate.containBlock(msg.getBlockId())).thenReturn(false); + when(delegate.containBlock(msg.getBlockCapsule().getParentBlockId())).thenReturn(true); + when(delegate.validBlock(any(BlockCapsule.class))).thenReturn(true); + when(delegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, HEAD_NUM)); + + double before = sample(); + request(msg); + handler.processMessage(peer, msg); + assertEquals(before, sample(), 0.0); + // the response-time exact-id sample ran and found the block unknown + verify(delegate).containBlock(msg.getBlockId()); + // the unlink guard really did check the parent before the low-height branch ... + verify(delegate).containBlock(msg.getBlockCapsule().getParentBlockId()); + // ... which returned without delegating to block processing + verify(delegate, never()).processBlock(any(BlockCapsule.class), anyBoolean()); + } + + @Test + public void testNoRequestKnownBlockNotIncremented() throws Exception { + when(delegate.containBlock(any(BlockId.class))).thenReturn(true); + + double before = sample(); + try { + handler.processMessage(peer, newBlockMessage(1)); + fail("expected P2pException for a block with no matching request"); + } catch (P2pException e) { + assertEquals("no request", e.getMessage()); + } + assertEquals(before, sample(), 0.0); + } + + @Test + public void testMatchedRequestHeadDuplicateIncrements() throws P2pException { + // re-delivery of the current head block id counts: the id is exactly known even + // though it is not below head. + when(delegate.containBlock(any(BlockId.class))).thenReturn(true); + when(delegate.validBlock(any(BlockCapsule.class))).thenReturn(true); + when(delegate.getHeadBlockId()).thenReturn(new BlockId(Sha256Hash.ZERO_HASH, HEAD_NUM)); + when(delegate.getActivePeer()).thenReturn(Collections.emptyList()); + + BlockMessage msg = newBlockMessage(HEAD_NUM); + double before = sample(); + request(msg); + handler.processMessage(peer, msg); + assertEquals(before + 1, sample(), 0.0); + } + + private BlockMessage newBlockMessage(long number) { + BlockCapsule capsule = new BlockCapsule(number, Sha256Hash.ZERO_HASH, + System.currentTimeMillis() - 60_000L, Sha256Hash.ZERO_HASH.getByteString()); + return new BlockMessage(capsule); + } + + private void request(BlockMessage msg) { + peer.getAdvInvRequest() + .put(new Item(msg.getBlockId(), InventoryType.BLOCK), System.currentTimeMillis()); + } + + private double sample() { + Double value = CollectorRegistry.defaultRegistry.getSampleValue(SAMPLE_NAME); + return value == null ? 0 : value; + } +} diff --git a/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java b/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java new file mode 100644 index 00000000000..b7a6ecdfcdc --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java @@ -0,0 +1,479 @@ +package org.tron.core.net.services; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.protobuf.ByteString; +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import org.junit.Assert; +import org.junit.Test; +import org.tron.common.BaseMethodTest; +import org.tron.common.utils.ReflectUtils; +import org.tron.common.utils.Sha256Hash; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.peer.Item; +import org.tron.core.net.peer.PeerConnection; +import org.tron.core.net.service.fetchblock.FetchBlockService; +import org.tron.p2p.connection.Channel; +import org.tron.protos.Protocol.Inventory.InventoryType; + +public class FetchBlockServiceTest extends BaseMethodTest { + + private FetchBlockService service; + private TronNetDelegate tronNetDelegate; + + @Override + protected void afterInit() { + service = context.getBean(FetchBlockService.class); + tronNetDelegate = mock(TronNetDelegate.class); + ReflectUtils.setFieldValue(service, "tronNetDelegate", tronNetDelegate); + } + + /** + * Verify that fetchBlockProcess selects the idle peer with the lowest avg latency + * (excluding the peer we are already fetching from) and sends it a FetchInvDataMessage. + * + *

Covers the migrated getPeerLatency / shouldFetchBlock path that replaced the + * legacy Dropwizard per-IP histogram P75 selection. + */ + @Test + public void testSelectLowestLatencyPeer() throws Exception { + InetSocketAddress oldAddr = new InetSocketAddress("127.0.0.1", 10001); + InetSocketAddress newAddr = new InetSocketAddress("127.0.0.2", 10001); + + Channel oldChannel = mock(Channel.class); + when(oldChannel.getInetSocketAddress()).thenReturn(oldAddr); + when(oldChannel.getInetAddress()).thenReturn(oldAddr.getAddress()); + when(oldChannel.getAvgLatency()).thenReturn(200L); + doNothing().when(oldChannel).send(any(byte[].class)); + + Channel newChannel = mock(Channel.class); + when(newChannel.getInetSocketAddress()).thenReturn(newAddr); + when(newChannel.getInetAddress()).thenReturn(newAddr.getAddress()); + when(newChannel.getAvgLatency()).thenReturn(50L); + doNothing().when(newChannel).send(any(byte[].class)); + + PeerConnection oldPeer = context.getBean(PeerConnection.class); + oldPeer.setChannel(oldChannel); + oldPeer.updateFetchLatency(200L); + PeerConnection newPeer = context.getBean(PeerConnection.class); + newPeer.setChannel(newChannel); + newPeer.updateFetchLatency(50L); + + Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Item item = new Item(hash, InventoryType.BLOCK); + + // both peers advertise having the block + oldPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); + newPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); + + List activePeers = new ArrayList<>(); + activePeers.add(oldPeer); + activePeers.add(newPeer); + when(tronNetDelegate.getActivePeer()).thenReturn(activePeers); + + // seed fetchBlockInfo via reflection (private static nested class) + Class fetchBlockInfoClass = Class.forName( + "org.tron.core.net.service.fetchblock.FetchBlockService$FetchBlockInfo"); + Constructor constructor = fetchBlockInfoClass.getDeclaredConstructor( + Sha256Hash.class, PeerConnection.class, long.class); + constructor.setAccessible(true); + Object fetchBlockInfo = constructor.newInstance( + hash, oldPeer, System.currentTimeMillis()); + ReflectUtils.setFieldValue(service, "fetchBlockInfo", fetchBlockInfo); + + // invoke private fetchBlockProcess + Method method = FetchBlockService.class.getDeclaredMethod( + "fetchBlockProcess", fetchBlockInfoClass); + method.setAccessible(true); + method.invoke(service, fetchBlockInfo); + + // new peer (lowest latency) should receive the fetch request + verify(newChannel).send(any(byte[].class)); + // old peer should not be re-requested + verify(oldChannel, never()).send(any(byte[].class)); + // fetchBlockInfo should be cleared after successful dispatch + Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + } + + /** + * A 600ms seed is clamped to 500ms, and >= timeout triggers fast-switch. + */ + @Test + public void testSwitchOnOldPeerTimeout() throws Exception { + InetSocketAddress oldAddr = new InetSocketAddress("127.0.0.3", 10001); + InetSocketAddress newAddr = new InetSocketAddress("127.0.0.4", 10001); + + Channel oldChannel = mock(Channel.class); + when(oldChannel.getInetSocketAddress()).thenReturn(oldAddr); + when(oldChannel.getInetAddress()).thenReturn(oldAddr.getAddress()); + // seed above timeout; estimator clamps this to default fetchBlockTimeout (500) + when(oldChannel.getAvgLatency()).thenReturn(600L); + doNothing().when(oldChannel).send(any(byte[].class)); + + Channel newChannel = mock(Channel.class); + when(newChannel.getInetSocketAddress()).thenReturn(newAddr); + when(newChannel.getInetAddress()).thenReturn(newAddr.getAddress()); + when(newChannel.getAvgLatency()).thenReturn(50L); + doNothing().when(newChannel).send(any(byte[].class)); + + PeerConnection oldPeer = context.getBean(PeerConnection.class); + ReflectUtils.setFieldValue(oldPeer, "channel", oldChannel); + oldPeer.updateFetchLatency(600L); + + PeerConnection newPeer = context.getBean(PeerConnection.class); + newPeer.setChannel(newChannel); + newPeer.updateFetchLatency(50L); + + Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Item item = new Item(hash, InventoryType.BLOCK); + + newPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); + + List activePeers = new ArrayList<>(); + activePeers.add(oldPeer); + activePeers.add(newPeer); + when(tronNetDelegate.getActivePeer()).thenReturn(activePeers); + + Class fetchBlockInfoClass = Class.forName( + "org.tron.core.net.service.fetchblock.FetchBlockService$FetchBlockInfo"); + Constructor constructor = fetchBlockInfoClass.getDeclaredConstructor( + Sha256Hash.class, PeerConnection.class, long.class); + constructor.setAccessible(true); + Object fetchBlockInfo = constructor.newInstance( + hash, oldPeer, System.currentTimeMillis()); + ReflectUtils.setFieldValue(service, "fetchBlockInfo", fetchBlockInfo); + + Method method = FetchBlockService.class.getDeclaredMethod( + "fetchBlockProcess", fetchBlockInfoClass); + method.setAccessible(true); + method.invoke(service, fetchBlockInfo); + + verify(newChannel).send(any(byte[].class)); + Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + } + + @Test + public void testSwitchOnHardTimeoutWhenOldPeerUnseeded() throws Exception { + PeerConnection oldPeer = context.getBean(PeerConnection.class); + PeerConnection candidate = context.getBean(PeerConnection.class); + Channel oldChannel = mock(Channel.class); + Channel candidateChannel = mock(Channel.class); + when(oldChannel.getAvgLatency()).thenReturn(0L); + when(candidateChannel.getAvgLatency()).thenReturn(50L); + ReflectUtils.setFieldValue(oldPeer, "channel", oldChannel); + ReflectUtils.setFieldValue(candidate, "channel", candidateChannel); + candidate.updateFetchLatency(50L); + doNothing().when(candidateChannel).send(any(byte[].class)); + + Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Item item = new Item(hash, InventoryType.BLOCK); + candidate.getAdvInvReceive().put(item, System.currentTimeMillis()); + when(tronNetDelegate.getActivePeer()).thenReturn(Arrays.asList(oldPeer, candidate)); + + Class fetchBlockInfoClass = Class.forName( + "org.tron.core.net.service.fetchblock.FetchBlockService$FetchBlockInfo"); + Constructor constructor = fetchBlockInfoClass.getDeclaredConstructor( + Sha256Hash.class, PeerConnection.class, long.class); + constructor.setAccessible(true); + Object fetchBlockInfo = constructor.newInstance( + hash, oldPeer, System.currentTimeMillis() - 600); + ReflectUtils.setFieldValue(service, "fetchBlockInfo", fetchBlockInfo); + Method method = FetchBlockService.class.getDeclaredMethod( + "fetchBlockProcess", fetchBlockInfoClass); + method.setAccessible(true); + method.invoke(service, fetchBlockInfo); + + verify(candidateChannel).send(any(byte[].class)); + Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + } + + /** + * Old peer unsampled: getFetchLatency() falls back to its channel avgLatency (0), while + * the candidate's first real sample 999 clamps to 500. shouldFetchBlock: left time + * 0 - 0 = 0, so 500 < 0 * 0.5 = 0 is false — no failover while the fetch is within timeout. + */ + @Test + public void testNoSwitchWhenOldPeerLatencyUnknown() throws Exception { + InetSocketAddress oldAddr = new InetSocketAddress("127.0.0.5", 10001); + InetSocketAddress newAddr = new InetSocketAddress("127.0.0.6", 10001); + + Channel oldChannel = mock(Channel.class); + when(oldChannel.getInetSocketAddress()).thenReturn(oldAddr); + when(oldChannel.getInetAddress()).thenReturn(oldAddr.getAddress()); + // old peer latency unknown + when(oldChannel.getAvgLatency()).thenReturn(0L); + doNothing().when(oldChannel).send(any(byte[].class)); + + Channel newChannel = mock(Channel.class); + when(newChannel.getInetSocketAddress()).thenReturn(newAddr); + when(newChannel.getInetAddress()).thenReturn(newAddr.getAddress()); + when(newChannel.getAvgLatency()).thenReturn(50L); + doNothing().when(newChannel).send(any(byte[].class)); + + PeerConnection oldPeer = context.getBean(PeerConnection.class); + oldPeer.setChannel(oldChannel); + + PeerConnection newPeer = context.getBean(PeerConnection.class); + newPeer.setChannel(newChannel); + // first real sample replaces the unsampled state, then clamps 999 to 500 + newPeer.updateFetchLatency(999L); + + Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Item item = new Item(hash, InventoryType.BLOCK); + + oldPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); + newPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); + + List activePeers = new ArrayList<>(); + activePeers.add(oldPeer); + activePeers.add(newPeer); + when(tronNetDelegate.getActivePeer()).thenReturn(activePeers); + + Class fetchBlockInfoClass = Class.forName( + "org.tron.core.net.service.fetchblock.FetchBlockService$FetchBlockInfo"); + Constructor constructor = fetchBlockInfoClass.getDeclaredConstructor( + Sha256Hash.class, PeerConnection.class, long.class); + constructor.setAccessible(true); + Object fetchBlockInfo = constructor.newInstance( + hash, oldPeer, System.currentTimeMillis()); + ReflectUtils.setFieldValue(service, "fetchBlockInfo", fetchBlockInfo); + + Method method = FetchBlockService.class.getDeclaredMethod( + "fetchBlockProcess", fetchBlockInfoClass); + method.setAccessible(true); + method.invoke(service, fetchBlockInfo); + + // no failover: candidate peer must not receive a fetch request + verify(newChannel, never()).send(any(byte[].class)); + // in-flight fetchBlockInfo stays pending + Assert.assertNotNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + } + + /** + * Both peers are unsampled, so both reads fall back to their channel avgLatency + * (old = 200, candidate = 0). The candidate wins min() and shouldFetchBlock: + * left time 200 - 0 = 200, so 0 < 200 * 0.5 = 100 holds — failover happens. + */ + @Test + public void testSwitchWhenBothPeersUnseededReadsChannelFallback() throws Exception { + InetSocketAddress oldAddr = new InetSocketAddress("127.0.0.7", 10001); + InetSocketAddress newAddr = new InetSocketAddress("127.0.0.8", 10001); + + Channel oldChannel = mock(Channel.class); + when(oldChannel.getInetSocketAddress()).thenReturn(oldAddr); + when(oldChannel.getInetAddress()).thenReturn(oldAddr.getAddress()); + when(oldChannel.getAvgLatency()).thenReturn(200L); + doNothing().when(oldChannel).send(any(byte[].class)); + + Channel newChannel = mock(Channel.class); + when(newChannel.getInetSocketAddress()).thenReturn(newAddr); + when(newChannel.getInetAddress()).thenReturn(newAddr.getAddress()); + // candidate unsampled: read falls back to its channel avgLatency (0) + when(newChannel.getAvgLatency()).thenReturn(0L); + doNothing().when(newChannel).send(any(byte[].class)); + + PeerConnection oldPeer = context.getBean(PeerConnection.class); + oldPeer.setChannel(oldChannel); + PeerConnection newPeer = context.getBean(PeerConnection.class); + newPeer.setChannel(newChannel); + + Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Item item = new Item(hash, InventoryType.BLOCK); + + oldPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); + newPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); + + List activePeers = new ArrayList<>(); + activePeers.add(oldPeer); + activePeers.add(newPeer); + when(tronNetDelegate.getActivePeer()).thenReturn(activePeers); + + Class fetchBlockInfoClass = Class.forName( + "org.tron.core.net.service.fetchblock.FetchBlockService$FetchBlockInfo"); + Constructor constructor = fetchBlockInfoClass.getDeclaredConstructor( + Sha256Hash.class, PeerConnection.class, long.class); + constructor.setAccessible(true); + Object fetchBlockInfo = constructor.newInstance( + hash, oldPeer, System.currentTimeMillis()); + ReflectUtils.setFieldValue(service, "fetchBlockInfo", fetchBlockInfo); + + Method method = FetchBlockService.class.getDeclaredMethod( + "fetchBlockProcess", fetchBlockInfoClass); + method.setAccessible(true); + method.invoke(service, fetchBlockInfo); + + // failover: unsampled candidate reads channel fallback 0 and wins the comparison + verify(newChannel).send(any(byte[].class)); + Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + } + + /** + * Old peer seeded at 200; the unsampled candidate reads its channel fallback (0), wins + * min() and the left-time comparison (0 < 200 * 0.5) — failover to the candidate. + */ + @Test + public void testSwitchToUnseededCandidateWhenOldPeerSeeded() throws Exception { + PeerConnection oldPeer = context.getBean(PeerConnection.class); + PeerConnection candidate = context.getBean(PeerConnection.class); + Channel oldChannel = mock(Channel.class); + Channel candidateChannel = mock(Channel.class); + when(oldChannel.getAvgLatency()).thenReturn(200L); + when(candidateChannel.getAvgLatency()).thenReturn(0L); + ReflectUtils.setFieldValue(oldPeer, "channel", oldChannel); + oldPeer.updateFetchLatency(200L); + ReflectUtils.setFieldValue(candidate, "channel", candidateChannel); + doNothing().when(candidateChannel).send(any(byte[].class)); + + Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Item item = new Item(hash, InventoryType.BLOCK); + candidate.getAdvInvReceive().put(item, System.currentTimeMillis()); + when(tronNetDelegate.getActivePeer()).thenReturn(Arrays.asList(oldPeer, candidate)); + + Class infoClass = Class.forName( + "org.tron.core.net.service.fetchblock.FetchBlockService$FetchBlockInfo"); + Constructor constructor = infoClass.getDeclaredConstructor( + Sha256Hash.class, PeerConnection.class, long.class); + constructor.setAccessible(true); + Object info = constructor.newInstance(hash, oldPeer, System.currentTimeMillis()); + ReflectUtils.setFieldValue(service, "fetchBlockInfo", info); + Method method = FetchBlockService.class.getDeclaredMethod("fetchBlockProcess", infoClass); + method.setAccessible(true); + method.invoke(service, info); + + verify(candidateChannel).send(any(byte[].class)); + Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + } + + /** + * The first real fetch sample directly replaces the unsampled state (no blending with + * the channel prior, isomorphic to RFC 6298 SRTT initialization); the clamp still + * applies, so 999 saturates to the 500ms bound instead of the old blended (123*9+999)/10. + */ + @Test + public void testFirstFetchLatencySampleReplacesSeed() { + Channel channel = mock(Channel.class); + when(channel.getAvgLatency()).thenReturn(123L); + PeerConnection peer = new PeerConnection(); + ReflectUtils.setFieldValue(peer, "channel", channel); + + // First real sample replaces the unsampled state; channel prior (123) is not blended. + peer.updateFetchLatency(999L); + + Assert.assertEquals(500L, peer.getFetchLatency()); + } + + @Test + public void testFetchLatencyUsesEwma() { + Channel channel = mock(Channel.class); + when(channel.getAvgLatency()).thenReturn(100L); + PeerConnection peer = new PeerConnection(); + ReflectUtils.setFieldValue(peer, "channel", channel); + + // first real sample initializes directly: clamp(100) = 100 + peer.updateFetchLatency(100L); + // second sample onwards: EWMA alpha = 0.1 → (100 * 9 + 200) / 10 = 110 + peer.updateFetchLatency(200L); + + Assert.assertEquals(110L, peer.getFetchLatency()); + } + + /** + * Degradation: seeded at 100 (direct replacement), then ten 500ms samples. With + * alpha = 0.1 and integer division the estimate rises monotonically while converging + * smoothly and never exceeds the 500ms clamp bound. Hand-computed sequence: + * 140, 176, 208, 237, 263, 286, 307, 326, 343, 358. + */ + @Test + public void testFetchLatencyEwmaDegradationConverges() { + Channel channel = mock(Channel.class); + when(channel.getAvgLatency()).thenReturn(0L); + PeerConnection peer = new PeerConnection(); + ReflectUtils.setFieldValue(peer, "channel", channel); + + // first real sample initializes directly: clamp(100) = 100 + peer.updateFetchLatency(100L); + long previous = peer.getFetchLatency(); + long[] expected = {140, 176, 208, 237, 263, 286, 307, 326, 343, 358}; + for (long expectedValue : expected) { + peer.updateFetchLatency(500L); + long current = peer.getFetchLatency(); + Assert.assertEquals(expectedValue, current); + Assert.assertTrue(current > previous); + Assert.assertTrue(current <= 500L); + previous = current; + } + } + + /** + * Recovery: continuing from the degradation endpoint (358), ten 100ms samples pull the + * estimate back down monotonically and smoothly. Hand-computed sequence: + * 332, 308, 287, 268, 251, 235, 221, 208, 197, 187. + */ + @Test + public void testFetchLatencyEwmaRecoveryConverges() { + Channel channel = mock(Channel.class); + when(channel.getAvgLatency()).thenReturn(0L); + PeerConnection peer = new PeerConnection(); + ReflectUtils.setFieldValue(peer, "channel", channel); + + // replay the degradation phase to reach its endpoint + peer.updateFetchLatency(100L); + for (int i = 0; i < 10; i++) { + peer.updateFetchLatency(500L); + } + Assert.assertEquals(358L, peer.getFetchLatency()); + + long previous = peer.getFetchLatency(); + long[] expected = {332, 308, 287, 268, 251, 235, 221, 208, 197, 187}; + for (long expectedValue : expected) { + peer.updateFetchLatency(100L); + long current = peer.getFetchLatency(); + Assert.assertEquals(expectedValue, current); + Assert.assertTrue(current < previous); + previous = current; + } + } + + @Test + public void testFetchLatencyIsClamped() { + Channel channel = mock(Channel.class); + when(channel.getAvgLatency()).thenReturn(100L); + PeerConnection peer = new PeerConnection(); + ReflectUtils.setFieldValue(peer, "channel", channel); + + // first real sample initializes directly: clamp(100) = 100 + peer.updateFetchLatency(100L); + // EWMA: (100 * 9 + 9999) / 10 = 1089, then clamped to the 500ms bound + peer.updateFetchLatency(9999L); + + Assert.assertEquals(500L, peer.getFetchLatency()); + } + + @Test + public void testFetchLatencyIsIsolatedAcrossConnections() { + Channel firstChannel = mock(Channel.class); + when(firstChannel.getAvgLatency()).thenReturn(100L); + PeerConnection first = new PeerConnection(); + ReflectUtils.setFieldValue(first, "channel", firstChannel); + first.updateFetchLatency(9999L); + + Channel secondChannel = mock(Channel.class); + when(secondChannel.getAvgLatency()).thenReturn(50L); + PeerConnection second = new PeerConnection(); + ReflectUtils.setFieldValue(second, "channel", secondChannel); + + // the fresh connection is unsampled: it reads its own channel fallback (50), + // not the first connection's estimate (500) and not a cross-connection value + Assert.assertEquals(50L, second.getFetchLatency()); + } +} diff --git a/protocol/src/main/protos/api/api.proto b/protocol/src/main/protos/api/api.proto index f8d13a6bbd3..f2f51392b65 100644 --- a/protocol/src/main/protos/api/api.proto +++ b/protocol/src/main/protos/api/api.proto @@ -646,6 +646,8 @@ service Database { }; service Monitor { + option deprecated = true; + rpc GetStatsInfo (EmptyMessage) returns (MetricsInfo) { } } diff --git a/protocol/src/main/protos/core/Tron.proto b/protocol/src/main/protos/core/Tron.proto index a68e841bb60..50046903a5b 100644 --- a/protocol/src/main/protos/core/Tron.proto +++ b/protocol/src/main/protos/core/Tron.proto @@ -756,6 +756,8 @@ message NodeInfo { } message MetricsInfo { + option deprecated = true; + int64 interval = 1; NodeInfo node = 2; BlockChainInfo blockchain = 3;