diff --git a/framework/src/main/java/org/tron/core/net/P2pEventHandlerImpl.java b/framework/src/main/java/org/tron/core/net/P2pEventHandlerImpl.java index dda4f8d5f02..bbebd0c5dea 100644 --- a/framework/src/main/java/org/tron/core/net/P2pEventHandlerImpl.java +++ b/framework/src/main/java/org/tron/core/net/P2pEventHandlerImpl.java @@ -259,7 +259,6 @@ private void updateLastInteractiveTime(PeerConnection peer, TronMessage msg) { switch (type) { case SYNC_BLOCK_CHAIN: case BLOCK_CHAIN_INVENTORY: - case BLOCK: flag = true; break; case FETCH_INV_DATA: @@ -269,7 +268,7 @@ private void updateLastInteractiveTime(PeerConnection peer, TronMessage msg) { break; } if (flag) { - peer.setLastInteractiveTime(System.currentTimeMillis()); + peer.updateLastInteractiveTime(System.currentTimeMillis()); } } diff --git a/framework/src/main/java/org/tron/core/net/TronNetDelegate.java b/framework/src/main/java/org/tron/core/net/TronNetDelegate.java index 4b5524c8eb0..5694a7c1f20 100644 --- a/framework/src/main/java/org/tron/core/net/TronNetDelegate.java +++ b/framework/src/main/java/org/tron/core/net/TronNetDelegate.java @@ -59,6 +59,7 @@ import org.tron.core.net.message.MessageTypes; import org.tron.core.net.message.adv.BlockMessage; import org.tron.core.net.message.adv.TransactionMessage; +import org.tron.core.net.peer.Item; import org.tron.core.net.peer.PeerConnection; import org.tron.core.store.WitnessScheduleStore; import org.tron.protos.Protocol.Inventory.InventoryType; @@ -138,6 +139,35 @@ public Collection getActivePeer() { return TronNetService.getPeers(); } + public void recordInventory(PeerConnection peer, Item item, long receivedAt) { + if (item.getType() != InventoryType.BLOCK) { + peer.getAdvInvReceive().put(item, receivedAt); + return; + } + synchronized (this) { + // Capture eligibility before fetching; validation may advance the head immediately. + if (peer.getAdvInvSpread().getIfPresent(item) == null + && new BlockId(item.getHash()).getNum() > getHeadBlockId().getNum()) { + peer.getAdvBlockInvReceive().put(item, receivedAt); + } + peer.getAdvInvReceive().put(item, receivedAt); + } + } + + /** + * Confirms eligible announcements only after the block has been successfully processed. + */ + public synchronized void confirmBlockInventory(BlockId blockId) { + Item item = new Item(blockId, InventoryType.BLOCK); + // Share the registration lock so confirmation cannot miss an eligible announcement. + getActivePeer().forEach(peer -> { + Long receivedAt = peer.getAdvBlockInvReceive().asMap().remove(item); + if (receivedAt != null) { + peer.updateLastInteractiveTime(receivedAt); + } + }); + } + public long getSyncBeginNumber() { return dbManager.getSyncBeginNumber(); } 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 b1631979913..08358a5dcce 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 @@ -9,6 +9,7 @@ import org.springframework.stereotype.Component; import org.tron.common.prometheus.MetricKeys; import org.tron.common.prometheus.Metrics; +import org.tron.common.utils.Sha256Hash; import org.tron.core.Constant; import org.tron.core.capsule.BlockCapsule; import org.tron.core.capsule.BlockCapsule.BlockId; @@ -71,9 +72,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep blockMessage.getBlockId(), peer.getInetSocketAddress()); throw new P2pException(TypeEnum.BAD_MESSAGE, "block time error"); } - if (!fastForward && !peer.isRelayPeer()) { - check(peer, blockMessage); - } + check(peer, blockMessage); blockMessage.sanitize(); @@ -87,27 +86,22 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep if (peer.isRelayPeer()) { peer.getAdvInvSpread().put(item, now); } - Long time = peer.getAdvInvRequest().remove(item); + Long time = peer.getAdvInvRequest().get(item); if (null != 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); - fetchBlockService.blockFetchSuccess(blockId); long interval = blockId.getNum() - tronNetDelegate.getHeadBlockId().getNum(); - processBlock(peer, blockMessage.getBlockCapsule()); + BlockResult result = processBlock(peer, blockMessage.getBlockCapsule()); + if (result == BlockResult.ACCEPTED || result == BlockResult.SYNC_REQUIRED) { + peer.setBlockRcvTime(System.currentTimeMillis()); + } logger.info( "Receive block/interval {}/{} from {} fetch/delay {}/{}ms, " + "txs/process {}/{}ms, witness: {}", @@ -123,55 +117,77 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep } private void check(PeerConnection peer, BlockMessage msg) throws P2pException { - Item item = new Item(msg.getBlockId(), InventoryType.BLOCK); - if (!peer.getSyncBlockRequested().containsKey(msg.getBlockId()) && !peer.getAdvInvRequest() - .containsKey(item)) { - logger.error("Receive bad block {} from peer {}, with no request", - msg.getBlockId(), peer.getInetSocketAddress()); - throw new P2pException(TypeEnum.BAD_MESSAGE, "no request"); + if (!fastForward && !peer.isRelayPeer()) { + Item item = new Item(msg.getBlockId(), InventoryType.BLOCK); + if (!peer.getSyncBlockRequested().containsKey(msg.getBlockId()) && !peer.getAdvInvRequest() + .containsKey(item)) { + logger.error("Receive bad block {} from peer {}, with no request", + msg.getBlockId(), peer.getInetSocketAddress()); + throw new P2pException(TypeEnum.BAD_MESSAGE, "no request"); + } + } + + BlockCapsule blockCapsule = msg.getBlockCapsule(); + if (blockCapsule.getNum() <= 0 + || blockCapsule.getParentHashStr().size() != Sha256Hash.LENGTH + || new BlockId(blockCapsule.getParentHash()).getNum() != blockCapsule.getNum() - 1) { + throw new P2pException(TypeEnum.BAD_BLOCK, "block number does not follow parent"); } } - private void processBlock(PeerConnection peer, BlockCapsule block) throws P2pException { + private BlockResult processBlock(PeerConnection peer, BlockCapsule block) throws P2pException { BlockId blockId = block.getBlockId(); - boolean flag = tronNetDelegate.validBlock(block); - if (!flag) { - logger.warn("Receive a bad block from {}, {}, {}", + boolean activeWitness = tronNetDelegate.validBlock(block); + // Retain pending fetch/request state if validation throws so disconnect can retry. + // Otherwise, complete the fetch. + fetchBlockService.blockFetchSuccess(blockId, peer); + if (!activeWitness) { + logger.warn("Receive a block from an inactive witness, peer {}, block {}, witness {}", peer.getInetSocketAddress(), blockId.getString(), Hex.toHexString(block.getWitnessAddress().toByteArray())); - return; + return BlockResult.STATE_FAILED; + } + + peer.updateLastInteractiveTime(System.currentTimeMillis()); + long headNum = tronNetDelegate.getHeadBlockId().getNum(); + if (block.getNum() < headNum || tronNetDelegate.containBlock(blockId)) { + logger.warn("Receive a low block {}, head {}", blockId.getString(), headNum); + return BlockResult.IGNORED; } if (!tronNetDelegate.containBlock(block.getParentBlockId())) { logger.warn("Get unlink block {} from {}, head is {}", blockId.getString(), peer.getInetAddress(), tronNetDelegate.getHeadBlockId().getString()); syncService.startSync(peer); - return; - } - - long headNum = tronNetDelegate.getHeadBlockId().getNum(); - if (block.getNum() < headNum) { - logger.warn("Receive a low block {}, head {}", blockId.getString(), headNum); - return; + return BlockResult.SYNC_REQUIRED; } broadcast(new BlockMessage(block)); try { tronNetDelegate.processBlock(block, false); - peer.setBlockRcvTime(System.currentTimeMillis()); - witnessProductBlockService.validWitnessProductTwoBlock(block); - - Item item = new Item(blockId, InventoryType.BLOCK); - tronNetDelegate.getActivePeer().forEach(p -> { - if (p.getAdvInvReceive().getIfPresent(item) != null) { - p.setBlockBothHave(blockId); - } - }); } catch (Exception e) { logger.warn("Process adv block {} from peer {} failed. reason: {}", blockId, peer.getInetAddress(), e.getMessage()); + return BlockResult.STATE_FAILED; } + if (tronNetDelegate.isHitDown()) { + return BlockResult.IGNORED; + } + tronNetDelegate.confirmBlockInventory(blockId); + witnessProductBlockService.validWitnessProductTwoBlock(block); + + Item item = new Item(blockId, InventoryType.BLOCK); + tronNetDelegate.getActivePeer().forEach(p -> { + if (p.getAdvInvReceive().getIfPresent(item) != null) { + p.setBlockBothHave(blockId); + } + }); + return BlockResult.ACCEPTED; + } + + private enum BlockResult { + ACCEPTED, SYNC_REQUIRED, IGNORED, STATE_FAILED } private void broadcast(BlockMessage blockMessage) { diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java index 33cf32462f1..2c0bf1b9209 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java @@ -52,6 +52,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep if (blockIdWeGet.size() == 1 && tronNetDelegate.containBlock(blockIdWeGet.peek())) { peer.setTronState(TronState.SYNC_COMPLETED); peer.setNeedSyncFromPeer(false); + syncService.setHandleFlag(true); return; } @@ -82,6 +83,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep } peer.setFetchAble(true); + syncService.setHandleFlag(true); if ((chainInventoryMessage.getRemainNum() == 0 && !peer.getSyncBlockToFetch().isEmpty()) || (chainInventoryMessage.getRemainNum() != 0 && peer.getSyncBlockToFetch().size() > syncFetchBatchNum)) { diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java index f96f7f0b0ff..8bece5400cd 100644 --- a/framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java +++ b/framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java @@ -6,7 +6,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.tron.common.utils.Sha256Hash; -import org.tron.core.capsule.BlockCapsule.BlockId; import org.tron.core.config.args.Args; import org.tron.core.exception.P2pException; import org.tron.core.exception.P2pException.TypeEnum; @@ -42,14 +41,8 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep for (Sha256Hash id : inventoryMessage.getHashList()) { Item item = new Item(id, type); - peer.getAdvInvReceive().put(item, System.currentTimeMillis()); + tronNetDelegate.recordInventory(peer, item, System.currentTimeMillis()); advService.addInv(item); - if (type.equals(InventoryType.BLOCK) && peer.getAdvInvSpread().getIfPresent(item) == null) { - long headNum = tronNetDelegate.getHeadBlockId().getNum(); - if (new BlockId(id).getNum() > headNum) { - peer.setLastInteractiveTime(System.currentTimeMillis()); - } - } } } 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 91c3b145556..94c1ab3bf5b 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 @@ -106,8 +106,8 @@ public class PeerConnection { * clamped to fetchBlock.timeout only afterwards, so sustained degradation saturates * within 1-3 samples (issue #6923). Accepted cost: one transient outlier (~10x a * healthy estimate) can jump the estimate straight to the clamp, needing ~9 healthy - * samples to decay; hard-timeout fallback and the saturation gate (one secondary - * fetch per tick) bound the impact meanwhile. + * samples to decay; backup dispatch still requires an elapsed timeout or sufficient + * expected benefit. */ private static final int FETCH_LATENCY_EWMA_FACTOR = 10; @@ -141,6 +141,11 @@ public class PeerConnection { private Cache advInvReceive = CacheBuilder.newBuilder().maximumSize(INV_CACHE_SIZE) .expireAfterWrite(1, TimeUnit.HOURS).recordStats().build(); + // Eligible block INV receipt times, pending successful block processing before activity updates. + @Getter + private Cache advBlockInvReceive = CacheBuilder.newBuilder().maximumSize(100) + .expireAfterWrite(1, TimeUnit.MINUTES).build(); + @Setter @Getter private Cache advInvSpread = CacheBuilder.newBuilder().maximumSize(INV_CACHE_SIZE) @@ -195,7 +200,7 @@ public void setChannel(Channel channel) { this.isRelayPeer = true; } this.nodeStatistics = TronStatsManager.getNodeStatistics(channel.getInetAddress()); - lastInteractiveTime = System.currentTimeMillis(); + updateLastInteractiveTime(System.currentTimeMillis()); p2pRateLimiter.register(SYNC_BLOCK_CHAIN.asByte(), Args.getInstance().getRateLimiterSyncBlockChain()); p2pRateLimiter.register(FETCH_INV_DATA.asByte(), @@ -238,7 +243,7 @@ public long getFetchLatency() { } private long clampFetchLatency(long latency) { - // Saturation intentionally makes the >= timeout gate in FetchBlockService trigger. + // Bound measured latency without treating saturation as an elapsed fetch timeout. return StrictMathWrapper.max(0, StrictMathWrapper.min(CommonParameter.getInstance().fetchBlockTimeout, latency)); } @@ -248,10 +253,39 @@ public void setBlockBothHave(BlockId blockId) { this.blockBothHaveUpdateTime = System.currentTimeMillis(); } + public synchronized void updateLastInteractiveTime(long time) { + if (time > lastInteractiveTime) { + lastInteractiveTime = time; + } + } + + /** + * Returns whether there are no outstanding inventory or sync requests. + * + *

Sync blocks already being processed and sync direction flags are not checked. + */ public boolean isIdle() { return advInvRequest.isEmpty() && isSyncIdle(); } + /** + * Returns whether there are no outstanding block or sync requests and no sync blocks + * being processed. Outstanding transaction inventory requests do not make this check fail. + * + *

Callers must check connection state and sync direction flags separately before fetching. + */ + public boolean isBlockFetchIdle() { + return advInvRequest.keySet().stream() + .noneMatch(item -> item.getType() == Protocol.Inventory.InventoryType.BLOCK) + && isSyncIdle() && syncBlockInProcess.isEmpty(); + } + + /** + * Returns whether there are no outstanding sync block or chain-summary requests. + * + *

This does not mean synchronization is complete: received blocks may still be processing, + * and sync direction flags are not checked. + */ public boolean isSyncIdle() { return syncBlockRequested.isEmpty() && syncChainRequested == null; } @@ -284,6 +318,7 @@ public void onDisconnect() { syncService.onDisconnect(this); advService.onDisconnect(this); advInvReceive.invalidateAll(); + advBlockInvReceive.invalidateAll(); advInvSpread.invalidateAll(); advInvRequest.clear(); syncBlockIdCache.invalidateAll(); diff --git a/framework/src/main/java/org/tron/core/net/service/adv/AdvService.java b/framework/src/main/java/org/tron/core/net/service/adv/AdvService.java index 2241ce8d1ce..672a43b207a 100644 --- a/framework/src/main/java/org/tron/core/net/service/adv/AdvService.java +++ b/framework/src/main/java/org/tron/core/net/service/adv/AdvService.java @@ -8,11 +8,13 @@ import com.google.common.cache.CacheBuilder; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map.Entry; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -26,6 +28,7 @@ import org.tron.common.utils.Sha256Hash; import org.tron.common.utils.Time; import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.config.Parameter.NetConstants; import org.tron.core.config.args.Args; import org.tron.core.net.TronNetDelegate; import org.tron.core.net.message.adv.BlockMessage; @@ -41,6 +44,7 @@ @Slf4j(topic = "net") @Component public class AdvService { + private final int MAX_INV_TO_FETCH_CACHE_SIZE = 100_000; private final int MAX_TRX_CACHE_SIZE = 50_000; private final int MAX_BLOCK_CACHE_SIZE = 10; @@ -159,8 +163,8 @@ public Message getMessage(Item item) { public int fastBroadcastTransaction(TransactionMessage msg) { List peers = tronNetDelegate.getActivePeer().stream() - .filter(peer -> !peer.isNeedSyncFromPeer() && !peer.isNeedSyncFromUs()) - .collect(Collectors.toList()); + .filter(peer -> !peer.isNeedSyncFromPeer() && !peer.isNeedSyncFromUs()) + .collect(Collectors.toList()); if (peers.size() == 0) { logger.warn("Broadcast transaction {} failed, no connection", msg.getMessageId()); @@ -176,9 +180,9 @@ public int fastBroadcastTransaction(TransactionMessage msg) { InventoryMessage inventoryMessage = new InventoryMessage(list, InventoryType.TRX); int peersCount = 0; - for (PeerConnection peer: peers) { + for (PeerConnection peer : peers) { if (peer.getAdvInvReceive().getIfPresent(item) == null - && peer.getAdvInvSpread().getIfPresent(item) == null) { + && peer.getAdvInvSpread().getIfPresent(item) == null) { peersCount++; peer.getAdvInvSpread().put(item, Time.getCurrentMillis()); peer.sendMessage(inventoryMessage); @@ -229,75 +233,117 @@ public void broadcast(Message msg) { } } - /* - public void fastForward(BlockMessage msg) { - Item item = new Item(msg.getBlockId(), InventoryType.BLOCK); - List peers = tronNetDelegate.getActivePeer().stream() - .filter(peer -> !peer.isNeedSyncFromPeer() && !peer.isNeedSyncFromUs()) - .filter(peer -> peer.getAdvInvReceive().getIfPresent(item) == null - && peer.getAdvInvSpread().getIfPresent(item) == null) - .collect(Collectors.toList()); + public void onDisconnect(PeerConnection peer) { + // Release this peer's retry tracking before looking for another outstanding request. + fetchBlockService.onDisconnect(peer); + peer.getAdvInvRequest().keySet().forEach(item -> recoverFetchOnDisconnect(peer, item)); - if (!fastForward) { - peers = peers.stream().filter(peer -> peer.isFastForwardPeer()).collect(Collectors.toList()); + if (!invToFetch.isEmpty()) { + consumerInvToFetch(); } - - peers.forEach(peer -> { - peer.fastSend(msg); - peer.getAdvInvSpread().put(item, System.currentTimeMillis()); - peer.setFastForwardBlock(msg.getBlockId()); - }); } - */ - public void onDisconnect(PeerConnection peer) { - if (!peer.getAdvInvRequest().isEmpty()) { - peer.getAdvInvRequest().keySet().forEach(item -> { - if (tronNetDelegate.getActivePeer().stream() - .anyMatch(p -> !p.equals(peer) && p.getAdvInvReceive().getIfPresent(item) != null)) { - invToFetch.put(item, System.currentTimeMillis()); - } else { - invToFetchCache.invalidate(item); - } - }); + private synchronized void recoverFetchOnDisconnect(PeerConnection peer, Item item) { + Collection peers = tronNetDelegate.getActivePeer().stream() + .filter(p -> !p.equals(peer) && !p.isDisconnect()).collect(Collectors.toList()); + // Another provider may have already delivered the block. + if (item.getType() == InventoryType.BLOCK + && (blockCache.getIfPresent(item) != null + || tronNetDelegate.containBlock(new BlockId(item.getHash())))) { + return; } - - if (invToFetch.size() > 0) { - consumerInvToFetch(); + if (item.getType() == InventoryType.BLOCK) { + Optional pending = peers.stream() + .filter(p -> p.getAdvInvRequest().containsKey(item)).findFirst(); + if (pending.isPresent()) { + // Reuse the pending request and its original timestamp instead of sending it again. + fetchBlockService.fetchBlock(Collections.singletonList(item.getHash()), pending.get()); + return; + } + } + if (peers.stream().anyMatch(p -> p.getAdvInvReceive().getIfPresent(item) != null)) { + // Let normal scheduling select a replacement from the remaining advertisers. + invToFetch.put(item, System.currentTimeMillis()); + } else { + // Allow future announcements to enqueue the item again when no source remains. + invToFetch.remove(item); + invToFetchCache.invalidate(item); } } private void consumerInvToFetch() { - Collection peers = tronNetDelegate.getActivePeer().stream() - .filter(peer -> peer.isIdle()) + // Snapshot connected peers; block fetching can still use peers with pending TRX requests. + Collection peers = tronNetDelegate.getActivePeer(); + Collection trxPeers = peers.stream().filter(PeerConnection::isIdle) .collect(Collectors.toList()); + // Batch sizes count requests assigned in this pass, not all outstanding requests on a peer. InvSender invSender = new InvSender(); + // Coordinate queue updates with inventory admission, other consumers, and disconnect recovery. synchronized (this) { if (invToFetch.isEmpty() || peers.isEmpty()) { return; } long now = System.currentTimeMillis(); + List blocks = new ArrayList<>(); invToFetch.forEach((item, time) -> { - if (time < now - TIMEOUT) { + long timeout = item.getType() == InventoryType.BLOCK ? NetConstants.ADV_TIME_OUT : TIMEOUT; + if (time < now - timeout) { + // Release the deduplication entry so a later announcement can enqueue this item again. logger.info("This obj is too late to fetch, type: {} hash: {}", item.getType(), - item.getHash()); + item.getHash()); invToFetch.remove(item); invToFetchCache.invalidate(item); return; } - peers.stream().filter(peer -> { - Long t = peer.getAdvInvReceive().getIfPresent(item); - return t != null && now - t < TIMEOUT && invSender.getSize(peer) < MAX_TRX_FETCH_PER_PEER; - }).sorted(Comparator.comparingInt(peer -> invSender.getSize(peer))) - .findFirst().ifPresent(peer -> { + if (item.getType() == InventoryType.BLOCK) { + // Defer block allocation until all queued blocks can be ordered by height. + blocks.add(item); + return; + } + // Use recent advertisers, enforce the per-peer TRX batch limit, and prefer smaller batches. + trxPeers.stream() + .filter(peer -> { + Long t = peer.getAdvInvReceive().getIfPresent(item); + return t != null && now - t < TIMEOUT + && invSender.getSize(peer) < MAX_TRX_FETCH_PER_PEER; + }) + .min(Comparator.comparingInt(invSender::getSize)) + .ifPresent(peer -> { + // Reserve the request before sending; an existing request must not be sent again. if (peer.checkAndPutAdvInvRequest(item, now)) { invSender.add(item, peer); } invToFetch.remove(item); }); }); + + // Reserve peers for earlier blocks before later blocks can make them busy. + blocks.sort(Comparator.comparingLong(item -> new BlockId(item.getHash()).getNum())); + blocks.forEach(item -> { + // A cached, stored, or already requested block no longer needs an initial fetch. + if (blockCache.getIfPresent(item) != null + || tronNetDelegate.containBlock(new BlockId(item.getHash())) + || peers.stream().anyMatch(peer -> peer.getAdvInvRequest().containsKey(item))) { + invToFetch.remove(item); + return; + } + // Recheck eligibility after earlier reservations. Prefer fully idle providers, then + // smaller batches; providers with pending TRX requests remain eligible as a fallback. + peers.stream() + .filter(peer -> fetchBlockService.canFetchBlock(peer, item, now)) + .min(Comparator.comparing((PeerConnection peer) -> !peer.isIdle()) + .thenComparingInt(invSender::getSize)) + .ifPresent(peer -> { + if (peer.checkAndPutAdvInvRequest(item, now)) { + invSender.add(item, peer); + invToFetch.remove(item); + } + }); + }); + // Items without an eligible provider remain queued for a later pass. } + // Send outside the scheduling lock; sendFetch() registers block retry state before sending. invSender.sendFetch(); } @@ -327,7 +373,7 @@ private synchronized void consumerInvToSpread() { invSender.sendInv(); } - class InvSender { + private class InvSender { private HashMap>> send = new HashMap<>(); @@ -383,12 +429,12 @@ public void sendInv() { })); } - void sendFetch() { + private void sendFetch() { send.forEach((peer, ids) -> ids.forEach((key, value) -> { if (key.equals(InventoryType.BLOCK)) { value.sort(Comparator.comparingLong(value1 -> new BlockId(value1).getNum())); - peer.sendMessage(new FetchInvDataMessage(value, key)); fetchBlockService.fetchBlock(value, peer); + peer.sendMessage(new FetchInvDataMessage(value, key)); } else { peer.sendMessage(new FetchInvDataMessage(value, key)); } 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 d4a89dc846d..fb074e29038 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 @@ -1,5 +1,6 @@ package org.tron.core.net.service.fetchblock; +import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.List; @@ -7,7 +8,6 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import lombok.Getter; -import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @@ -18,6 +18,7 @@ import org.tron.common.utils.Sha256Hash; import org.tron.core.ChainBaseManager; import org.tron.core.capsule.BlockCapsule; +import org.tron.core.config.Parameter.NetConstants; import org.tron.core.net.TronNetDelegate; import org.tron.core.net.message.adv.FetchInvDataMessage; import org.tron.core.net.peer.Item; @@ -34,12 +35,14 @@ public class FetchBlockService { @Autowired private ChainBaseManager chainBaseManager; - private volatile FetchBlockInfo fetchBlockInfo = null; + private volatile FetchBlockInfo fetchBlockInfo; private final long fetchTimeOut = CommonParameter.getInstance().fetchBlockTimeout; private static final double BLOCK_FETCH_LEFT_TIME_PERCENT = 0.5; + private static final int MAX_IN_FLIGHT_REQUESTS_PER_BLOCK = 2; + private final String esName = "fetch-block"; private final ScheduledExecutorService fetchBlockWorkerExecutor = @@ -59,83 +62,105 @@ public void close() { ExecutorServiceManager.shutdownAndAwaitTermination(fetchBlockWorkerExecutor, esName); } - public void fetchBlock(List sha256HashList, PeerConnection peer) { - if (sha256HashList.size() > 0) { + public synchronized void fetchBlock(List sha256HashList, PeerConnection peer) { + if (!sha256HashList.isEmpty()) { logger.info("Begin fetch block {} from {}", new BlockCapsule.BlockId(sha256HashList.get(0)).getString(), peer.getInetAddress()); } - if (null != fetchBlockInfo) { + long headNum = chainBaseManager.getHeadBlockNum(); + if (fetchBlockInfo != null + && new BlockCapsule.BlockId(fetchBlockInfo.getHash()).getNum() > headNum) { return; } - sha256HashList.stream().filter(sha256Hash -> new BlockCapsule.BlockId(sha256Hash).getNum() - == chainBaseManager.getHeadBlockNum() + 1) - .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); + fetchBlockInfo = null; + sha256HashList.stream() + .filter(sha256Hash -> new BlockCapsule.BlockId(sha256Hash).getNum() == headNum + 1) + .findFirst() + .ifPresent(sha256Hash -> { + Long requestTime = peer.getAdvInvRequest().get(new Item(sha256Hash, InventoryType.BLOCK)); + if (requestTime != null) { + fetchBlockInfo = new FetchBlockInfo(sha256Hash, peer, requestTime); + Metrics.counterInc(MetricKeys.Counter.BLOCK_FETCH_ARMED, 1); + logger.info("Set fetchBlockInfo, block: {}, peer: {}, time: {}", sha256Hash, + peer.getInetAddress(), requestTime); + } }); } - - public void blockFetchSuccess(Sha256Hash sha256Hash) { - logger.info("Fetch block success, {}", new BlockCapsule.BlockId(sha256Hash).getString()); + public synchronized void blockFetchSuccess(Sha256Hash sha256Hash, PeerConnection peer) { + peer.getAdvInvRequest().remove(new Item(sha256Hash, InventoryType.BLOCK)); FetchBlockInfo fetchBlockInfoTemp = this.fetchBlockInfo; if (null == fetchBlockInfoTemp || !fetchBlockInfoTemp.getHash().equals(sha256Hash)) { return; } + logger.info("Fetch block success, {}", new BlockCapsule.BlockId(sha256Hash).getString()); this.fetchBlockInfo = null; } - private void fetchBlockProcess(FetchBlockInfo fetchBlock) { - if (null == fetchBlock) { + public synchronized void onDisconnect(PeerConnection peer) { + if (fetchBlockInfo != null && fetchBlockInfo.getPeer().equals(peer)) { + fetchBlockInfo = null; + } + } + + public boolean canFetchBlock(PeerConnection peer, Item item, long now) { + if (peer.isDisconnect() || peer.isNeedSyncFromPeer() || peer.isNeedSyncFromUs() + || !peer.isBlockFetchIdle()) { + return false; + } + Long received = peer.getAdvInvReceive().getIfPresent(item); + return received != null && received >= now - NetConstants.ADV_TIME_OUT; + } + + private synchronized void fetchBlockProcess(FetchBlockInfo fetchBlock) { + if (fetchBlock == null || fetchBlockInfo != fetchBlock) { + return; + } + if (new BlockCapsule.BlockId(fetchBlock.getHash()).getNum() + <= chainBaseManager.getHeadBlockNum()) { + fetchBlockInfo = null; + return; + } + long now = System.currentTimeMillis(); + if (now - fetchBlock.getTime() >= NetConstants.ADV_TIME_OUT) { + // PeerStatusCheck owns the final deadline and disconnects the responsible provider. return; } Item item = new Item(fetchBlock.getHash(), InventoryType.BLOCK); - Optional optionalPeerConnection = tronNetDelegate.getActivePeer().stream() - .filter(PeerConnection::isIdle) - .filter(filterPeer -> !filterPeer.equals(fetchBlock.getPeer())) - .filter(filterPeer -> filterPeer.getAdvInvReceive().getIfPresent(item) != null) - // 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. + Collection peers = tronNetDelegate.getActivePeer(); + long pendingRequests = peers.stream() + .filter(peer -> peer.getAdvInvRequest().containsKey(item)) + .count(); + // Keep retry tracking and timeout responsibility while the original and backup are pending. + if (pendingRequests >= MAX_IN_FLIGHT_REQUESTS_PER_BLOCK) { + return; + } + Optional optionalPeerConnection = peers.stream() + .filter(peer -> !peer.equals(fetchBlock.getPeer())) + .filter(peer -> canFetchBlock(peer, item, now)) .min(Comparator.comparingDouble(this::getPeerLatency)); - if (optionalPeerConnection.isPresent()) { - optionalPeerConnection.ifPresent(firstPeer -> { - if (shouldFetchBlock(firstPeer, 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; - } - }); - } else { - if (System.currentTimeMillis() - fetchBlock.getTime() >= fetchTimeOut) { - logger.info("Clear fetchBlockInfo due to fetch block {} timeout {}ms", - fetchBlock.getHash(), fetchTimeOut); - this.fetchBlockInfo = null; + optionalPeerConnection.ifPresent(firstPeer -> { + if (shouldFetchBlock(firstPeer, fetchBlock) + && firstPeer.checkAndPutAdvInvRequest(item, now)) { + this.fetchBlockInfo = new FetchBlockInfo(item.getHash(), firstPeer, now); + firstPeer.sendMessage(new FetchInvDataMessage(Collections.singletonList(item.getHash()), + item.getType())); + Metrics.counterInc(MetricKeys.Counter.BLOCK_FETCH_SECONDARY, 1); } - } + }); } private boolean shouldFetchBlock(PeerConnection newPeer, FetchBlockInfo fetchBlock) { double newPeerLatency = getPeerLatency(newPeer); double oldPeerLatency = getPeerLatency(fetchBlock.getPeer()); long oldPeerSpendTime = System.currentTimeMillis() - fetchBlock.getTime(); - // Switch unconditionally on a hard timeout: an unseeded or saturated old peer must not - // permanently wedge fetchBlockInfo. + // After the actual request timeout, allow a backup without a useful latency estimate. if (oldPeerSpendTime >= fetchTimeOut) { return true; } - // 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; @@ -148,23 +173,18 @@ private double getPeerLatency(PeerConnection peerConnection) { private static class FetchBlockInfo { @Getter - @Setter - private PeerConnection peer; + private final PeerConnection peer; @Getter - @Setter - private Sha256Hash hash; + private final Sha256Hash hash; @Getter - @Setter - private long time; + private final long time; public FetchBlockInfo(Sha256Hash hash, PeerConnection peer, long time) { this.peer = peer; this.hash = hash; this.time = time; } - } - } diff --git a/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java b/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java index 75d1ec9f4b5..47ea62a3a06 100644 --- a/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java +++ b/framework/src/main/java/org/tron/core/net/service/sync/SyncService.java @@ -63,6 +63,8 @@ public class SyncService { private final ScheduledExecutorService blockHandleExecutor = ExecutorServiceManager .newSingleThreadScheduledExecutor(handleEsName); + // When true, the scheduled task calls handleSyncBlock() to process pending sync blocks. + @Setter private volatile boolean handleFlag = false; @Setter @@ -324,10 +326,19 @@ private synchronized void handleSyncBlock() { block = new BlockCapsule(unparsedBlock.getData()); } catch (Exception e) { logger.warn("Deserialize block {} failed", blockId.getString(), e); + peerConnection.getSyncBlockInProcess().remove(blockId); + invalid(blockId, peerConnection); return; } processSyncBlock(block, peerConnection); peerConnection.getSyncBlockInProcess().remove(blockId); + } else if (tronNetDelegate.getActivePeer().stream() + .noneMatch(peer -> peer.getSyncBlockToFetch().contains(blockId))) { + // No sync queue still needs this response; release it even if it is not local. + blockWaitToProcess.remove(unparsedBlock); + capacityReleased[0] = true; + peerConnection.getSyncBlockInProcess().remove(blockId); + invalid(blockId, peerConnection); } } }); @@ -344,8 +355,17 @@ private void processSyncBlock(BlockCapsule block, PeerConnection peerConnection) BlockId blockId = block.getBlockId(); try { tronNetDelegate.validSignature(block); + boolean useful = block.getNum() >= tronNetDelegate.getHeadBlockId().getNum() + && !tronNetDelegate.containBlock(blockId); tronNetDelegate.processBlock(block, true); - peerConnection.setBlockRcvTime(System.currentTimeMillis()); + if (tronNetDelegate.isHitDown()) { + return; + } + tronNetDelegate.confirmBlockInventory(blockId); + peerConnection.updateLastInteractiveTime(System.currentTimeMillis()); + if (useful) { + peerConnection.setBlockRcvTime(System.currentTimeMillis()); + } pbftDataSyncHandler.processPBFTCommitData(block); } catch (P2pException p2pException) { logger.error("Process sync block {} failed, type: {}", diff --git a/framework/src/test/java/org/tron/core/net/PeerBlockTestSupport.java b/framework/src/test/java/org/tron/core/net/PeerBlockTestSupport.java new file mode 100644 index 00000000000..b754b1df98f --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/PeerBlockTestSupport.java @@ -0,0 +1,44 @@ +package org.tron.core.net; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +import com.google.protobuf.ByteString; +import java.net.InetSocketAddress; +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.net.peer.PeerConnection; +import org.tron.p2p.connection.Channel; + +public final class PeerBlockTestSupport { + + private PeerBlockTestSupport() { + } + + public static PeerConnection peer(int port) { + PeerConnection peer = spy(new PeerConnection()); + Channel channel = mock(Channel.class); + InetSocketAddress address = new InetSocketAddress("127.0.0.1", port); + when(channel.getInetSocketAddress()).thenReturn(address); + when(channel.getInetAddress()).thenReturn(address.getAddress()); + ReflectUtils.setFieldValue(peer, "channel", channel); + peer.setNeedSyncFromPeer(false); + peer.setNeedSyncFromUs(false); + peer.setLastInteractiveTime(1L); + doNothing().when(peer).sendMessage(any()); + doNothing().when(peer).disconnect(any()); + return peer; + } + + public static BlockCapsule block(long number) { + BlockCapsule block = new BlockCapsule(number, new BlockId(Sha256Hash.ZERO_HASH, number - 1), + System.currentTimeMillis() - 1_000, ByteString.copyFromUtf8("witness")); + block.setMerkleRoot(); + return block; + } +} 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 index 64e77d77bba..1421e0ab478 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/BlockAlreadyKnownCounterTest.java @@ -1,7 +1,6 @@ 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; @@ -12,7 +11,6 @@ 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; @@ -108,14 +106,11 @@ public void testMatchedRequestAlreadyKnownBlockIncrements() throws P2pException @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 + // An unknown fork below head must not count. The low-height branch returns + // before checking the parent or delegating to block processing; 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)); @@ -125,9 +120,7 @@ public void testMatchedRequestUnknownForkBelowHeadNotIncremented() throws P2pExc 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 + // the low-height branch returned without delegating to block processing verify(delegate, never()).processBlock(any(BlockCapsule.class), anyBoolean()); } @@ -152,7 +145,6 @@ public void testMatchedRequestHeadDuplicateIncrements() 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)); - when(delegate.getActivePeer()).thenReturn(Collections.emptyList()); BlockMessage msg = newBlockMessage(HEAD_NUM); double before = sample(); @@ -162,7 +154,8 @@ public void testMatchedRequestHeadDuplicateIncrements() throws P2pException { } private BlockMessage newBlockMessage(long number) { - BlockCapsule capsule = new BlockCapsule(number, Sha256Hash.ZERO_HASH, + BlockCapsule capsule = new BlockCapsule(number, + new BlockId(Sha256Hash.ZERO_HASH, number - 1), System.currentTimeMillis() - 60_000L, Sha256Hash.ZERO_HASH.getByteString()); return new BlockMessage(capsule); } diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/BlockContributionTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/BlockContributionTest.java new file mode 100644 index 00000000000..93a706170ee --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/messagehandler/BlockContributionTest.java @@ -0,0 +1,345 @@ +package org.tron.core.net.messagehandler; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.doThrow; +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.Method; +import java.util.Arrays; +import java.util.Collections; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.tron.common.crypto.ECKey; +import org.tron.common.utils.ReflectUtils; +import org.tron.core.capsule.BlockCapsule; +import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.db.Manager; +import org.tron.core.exception.P2pException; +import org.tron.core.exception.P2pException.TypeEnum; +import org.tron.core.net.P2pEventHandlerImpl; +import org.tron.core.net.PeerBlockTestSupport; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.TronMessage; +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.sync.SyncService; +import org.tron.core.services.WitnessProductBlockService; +import org.tron.core.store.DynamicPropertiesStore; +import org.tron.core.store.WitnessScheduleStore; +import org.tron.protos.Protocol; +import org.tron.protos.Protocol.Inventory.InventoryType; + +public class BlockContributionTest { + + private BlockMsgHandler handler; + private TronNetDelegate delegate; + private FetchBlockService fetch; + private AdvService adv; + private SyncService sync; + private PeerConnection provider; + private BlockCapsule block; + private Item item; + private long requestTime; + + @Before + public void setUp() throws Exception { + handler = new BlockMsgHandler(); + delegate = mock(TronNetDelegate.class); + fetch = mock(FetchBlockService.class); + doCallRealMethod().when(fetch).blockFetchSuccess(any(), any()); + adv = mock(AdvService.class); + sync = mock(SyncService.class); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(handler, "fetchBlockService", fetch); + ReflectUtils.setFieldValue(handler, "advService", adv); + ReflectUtils.setFieldValue(handler, "syncService", sync); + ReflectUtils.setFieldValue(handler, "witnessProductBlockService", + mock(WitnessProductBlockService.class)); + ReflectUtils.setFieldValue(handler, "fastForward", false); + provider = PeerBlockTestSupport.peer(18888); + block = PeerBlockTestSupport.block(85636071); + item = new Item(block.getBlockId(), InventoryType.BLOCK); + requestTime = System.currentTimeMillis() - 1_000; + provider.getAdvInvRequest().put(item, requestTime); + when(delegate.getHeadBlockId()).thenReturn(block.getParentBlockId()); + when(delegate.getActivePeer()).thenReturn(Collections.singletonList(provider)); + when(delegate.containBlock(block.getParentBlockId())).thenReturn(true); + when(delegate.validBlock(block)).thenReturn(true); + } + + @Test + public void testValidationFetchBroadcastAndExecutionOrder() throws Exception { + when(delegate.validBlock(block)).thenAnswer(call -> { + Assert.assertEquals(Long.valueOf(requestTime), provider.getAdvInvRequest().get(item)); + verify(fetch, never()).blockFetchSuccess(any(), any()); + return true; + }); + doAnswer(call -> { + verify(fetch).blockFetchSuccess(block.getBlockId(), provider); + Assert.assertFalse(provider.getAdvInvRequest().containsKey(item)); + Assert.assertTrue(provider.getLastInteractiveTime() > 1); + Assert.assertEquals(0, provider.getBlockRcvTime()); + verify(delegate, never()).processBlock(any(), eq(false)); + return null; + }).when(adv).broadcast(any(BlockMessage.class)); + doAnswer(call -> { + verify(adv).broadcast(any(BlockMessage.class)); + Assert.assertEquals(0, provider.getBlockRcvTime()); + return null; + }).when(delegate).processBlock(block, false); + + handler.processMessage(provider, new BlockMessage(block)); + + Assert.assertTrue(provider.getBlockRcvTime() > 0); + } + + @Test + public void testOnlyProviderGetsContribution() throws Exception { + PeerConnection advertiser = PeerBlockTestSupport.peer(18889); + advertiser.getAdvInvReceive().put(item, System.currentTimeMillis()); + when(delegate.getActivePeer()).thenReturn(Arrays.asList(provider, advertiser)); + + handler.processMessage(provider, new BlockMessage(block)); + + Assert.assertTrue(provider.getBlockRcvTime() > 0); + Assert.assertEquals(0, advertiser.getBlockRcvTime()); + } + + @Test + public void testBadMerkleRetainsRequestAndDoesNotCompleteFetch() throws Exception { + assertInvalid(TypeEnum.BLOCK_MERKLE_INVALID); + } + + @Test + public void testBadSignatureRetainsRequestAndDoesNotCompleteFetch() throws Exception { + assertInvalid(TypeEnum.BLOCK_SIGN_INVALID); + } + + @Test + public void testInactiveWitnessDoesNotImproveEitherTimestamp() throws Exception { + when(delegate.validBlock(block)).thenReturn(false); + + handler.processMessage(provider, new BlockMessage(block)); + + Assert.assertEquals(1, provider.getLastInteractiveTime()); + Assert.assertEquals(0, provider.getBlockRcvTime()); + Assert.assertFalse(provider.getAdvInvRequest().containsKey(item)); + verify(sync, never()).startSync(any()); + verify(adv, never()).broadcast(any()); + verify(provider, never()).disconnect(any()); + } + + @Test + public void testMissingParentCreditsValidatedProviderAndStartsSync() throws Exception { + when(delegate.containBlock(block.getParentBlockId())).thenReturn(false); + + handler.processMessage(provider, new BlockMessage(block)); + + Assert.assertTrue(provider.getBlockRcvTime() > 0); + verify(sync).startSync(provider); + verify(adv, never()).broadcast(any()); + verify(delegate, never()).processBlock(any(), eq(false)); + } + + @Test + public void testOldOrphanDoesNotGetContribution() throws Exception { + when(delegate.getHeadBlockId()).thenReturn(new BlockId(block.getBlockId(), block.getNum() + 1)); + when(delegate.containBlock(block.getParentBlockId())).thenReturn(false); + + handler.processMessage(provider, new BlockMessage(block)); + + Assert.assertEquals(0, provider.getBlockRcvTime()); + verify(sync, never()).startSync(any()); + verify(adv, never()).broadcast(any()); + } + + @Test + public void testAlreadyKnownBlockDoesNotGetContribution() throws Exception { + when(delegate.containBlock(block.getBlockId())).thenReturn(true); + + handler.processMessage(provider, new BlockMessage(block)); + + Assert.assertEquals(0, provider.getBlockRcvTime()); + verify(adv, never()).broadcast(any()); + } + + @Test + public void testExecutionFailureDoesNotStartSyncOrDisconnectProvider() throws Exception { + doThrow(new P2pException(TypeEnum.BAD_BLOCK, "block execution failed")) + .when(delegate).processBlock(block, false); + + handler.processMessage(provider, new BlockMessage(block)); + + verify(adv).broadcast(any(BlockMessage.class)); + verify(sync, never()).startSync(any()); + verify(provider, never()).disconnect(any()); + Assert.assertTrue(provider.getLastInteractiveTime() > 1); + Assert.assertEquals(0, provider.getBlockRcvTime()); + } + + @Test + public void testShutdownDoesNotGetContribution() throws Exception { + when(delegate.isHitDown()).thenReturn(true); + + handler.processMessage(provider, new BlockMessage(block)); + + Assert.assertEquals(0, provider.getBlockRcvTime()); + } + + @Test + public void testUnrequestedBlockCannotImproveTimestamps() throws Exception { + provider.getAdvInvRequest().clear(); + try { + handler.processMessage(provider, new BlockMessage(block)); + Assert.fail("Expected an unrequested block to be rejected"); + } catch (P2pException e) { + Assert.assertEquals(TypeEnum.BAD_MESSAGE, e.getType()); + } + verify(delegate, never()).validBlock(any()); + Assert.assertEquals(1, provider.getLastInteractiveTime()); + Assert.assertEquals(0, provider.getBlockRcvTime()); + } + + @Test + public void testInvalidParentHeightCannotCompleteFetch() throws Exception { + BlockCapsule invalid = new BlockCapsule(block.getInstance().toBuilder() + .setBlockHeader(block.getInstance().getBlockHeader().toBuilder() + .setRawData(block.getInstance().getBlockHeader().getRawData().toBuilder() + .setNumber(block.getNum() + 1))).build()); + Item invalidItem = new Item(invalid.getBlockId(), InventoryType.BLOCK); + provider.getAdvInvRequest().put(invalidItem, requestTime); + try { + handler.processMessage(provider, new BlockMessage(invalid)); + Assert.fail("Expected invalid parent height to be rejected"); + } catch (P2pException e) { + Assert.assertEquals(TypeEnum.BAD_BLOCK, e.getType()); + } + Assert.assertEquals(Long.valueOf(requestTime), provider.getAdvInvRequest().get(invalidItem)); + verify(fetch, never()).blockFetchSuccess(any(), any()); + verify(adv, never()).broadcast(any()); + Assert.assertEquals(1, provider.getLastInteractiveTime()); + } + + @Test + public void testFastForwardAndRelayRejectInvalidParentHeight() throws Exception { + provider.getAdvInvRequest().clear(); + BlockCapsule invalid = new BlockCapsule(block.getInstance().toBuilder() + .setBlockHeader(block.getInstance().getBlockHeader().toBuilder() + .setRawData(block.getInstance().getBlockHeader().getRawData().toBuilder() + .setNumber(block.getNum() + 1))).build()); + + for (boolean fastForward : new boolean[] {true, false}) { + ReflectUtils.setFieldValue(handler, "fastForward", fastForward); + provider.setRelayPeer(!fastForward); + P2pException error = Assert.assertThrows(P2pException.class, + () -> handler.processMessage(provider, new BlockMessage(invalid))); + Assert.assertEquals(TypeEnum.BAD_BLOCK, error.getType()); + } + + verify(delegate, never()).validBlock(any()); + verify(fetch, never()).blockFetchSuccess(any(), any()); + verify(adv, never()).broadcast(any()); + Assert.assertEquals(1, provider.getLastInteractiveTime()); + Assert.assertEquals(0, provider.getBlockRcvTime()); + } + + @Test + public void testDispatcherDoesNotUnconditionallyCreditBlocks() throws Exception { + Method update = P2pEventHandlerImpl.class.getDeclaredMethod("updateLastInteractiveTime", + PeerConnection.class, TronMessage.class); + update.setAccessible(true); + + update.invoke(new P2pEventHandlerImpl(), provider, new BlockMessage(block)); + + Assert.assertEquals(1, provider.getLastInteractiveTime()); + } + + private void assertInvalid(TypeEnum type) throws Exception { + when(delegate.validBlock(block)).thenThrow(new P2pException(type, "invalid data")); + try { + handler.processMessage(provider, new BlockMessage(block)); + Assert.fail("Expected invalid block to be rejected"); + } catch (P2pException e) { + Assert.assertEquals(type, e.getType()); + } + Assert.assertEquals(Long.valueOf(requestTime), provider.getAdvInvRequest().get(item)); + Assert.assertEquals(1, provider.getLastInteractiveTime()); + Assert.assertEquals(0, provider.getBlockRcvTime()); + verify(fetch, never()).blockFetchSuccess(any(), any()); + verify(adv, never()).broadcast(any()); + } + + @Test(timeout = 20_000) + public void testRealSignedBlockPassesValidationBeforeContribution() throws Exception { + useRealValidation(); + handler.processMessage(provider, new BlockMessage(block)); + Assert.assertTrue(provider.getBlockRcvTime() > 0); + Assert.assertFalse(provider.getAdvInvRequest().containsKey(item)); + verify(fetch).blockFetchSuccess(block.getBlockId(), provider); + } + + @Test(timeout = 20_000) + public void testRealBadSignatureRetainsRequest() throws Exception { + useRealValidation(); + block = new BlockCapsule(block.getInstance().toBuilder() + .setBlockHeader(block.getInstance().getBlockHeader().toBuilder() + .setWitnessSignature(ByteString.copyFrom(new byte[65]))).build()); + assertRealValidationFailure(TypeEnum.BLOCK_SIGN_INVALID); + } + + @Test(timeout = 20_000) + public void testRealTamperedBodyRetainsRequest() throws Exception { + useRealValidation(); + block = new BlockCapsule(block.getInstance().toBuilder() + .addTransactions(Protocol.Transaction.newBuilder() + .setRawData(Protocol.Transaction.raw.newBuilder() + .setData(ByteString.copyFromUtf8("tampered body")))).build()); + assertRealValidationFailure(TypeEnum.BLOCK_MERKLE_INVALID); + } + + private void useRealValidation() throws Exception { + ECKey key = new ECKey(); + block = new BlockCapsule(block.getNum(), block.getParentBlockId(), block.getTimeStamp(), + ByteString.copyFrom(key.getAddress())); + block.setMerkleRoot(); + block.sign(key.getPrivKeyBytes()); + item = new Item(block.getBlockId(), InventoryType.BLOCK); + provider.getAdvInvRequest().clear(); + provider.getAdvInvRequest().put(item, requestTime); + + TronNetDelegate validator = new TronNetDelegate(); + Manager manager = mock(Manager.class); + when(manager.getDynamicPropertiesStore()).thenReturn(mock(DynamicPropertiesStore.class)); + WitnessScheduleStore witnesses = mock(WitnessScheduleStore.class); + when(witnesses.getActiveWitnesses()) + .thenReturn(Collections.singletonList(block.getWitnessAddress())); + ReflectUtils.setFieldValue(validator, "dbManager", manager); + ReflectUtils.setFieldValue(validator, "witnessScheduleStore", witnesses); + when(delegate.validBlock(any())).thenAnswer(call -> validator.validBlock(call.getArgument(0))); + } + + private void assertRealValidationFailure(TypeEnum type) throws Exception { + Assert.assertEquals(item.getHash(), block.getBlockId()); + try { + handler.processMessage(provider, new BlockMessage(block)); + Assert.fail("Expected cryptographic validation to reject the block"); + } catch (P2pException e) { + Assert.assertEquals(type, e.getType()); + } + Assert.assertEquals(Long.valueOf(requestTime), provider.getAdvInvRequest().get(item)); + Assert.assertEquals(1, provider.getLastInteractiveTime()); + Assert.assertEquals(0, provider.getBlockRcvTime()); + verify(fetch, never()).blockFetchSuccess(any(), any()); + } +} diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/BlockInventoryActivityTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/BlockInventoryActivityTest.java new file mode 100644 index 00000000000..95972c6bb1b --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/messagehandler/BlockInventoryActivityTest.java @@ -0,0 +1,415 @@ +package org.tron.core.net.messagehandler; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +import com.google.common.base.Ticker; +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.MockedStatic; +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.exception.P2pException.TypeEnum; +import org.tron.core.net.PeerBlockTestSupport; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.adv.BlockMessage; +import org.tron.core.net.message.adv.InventoryMessage; +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.sync.SyncService; +import org.tron.core.services.WitnessProductBlockService; +import org.tron.protos.Protocol.Inventory.InventoryType; + +public class BlockInventoryActivityTest { + + private AdvService adv; + private TronNetDelegate delegate; + private InventoryMsgHandler inventory; + private BlockMsgHandler blocks; + private PeerConnection provider; + private PeerConnection advertiser; + private BlockCapsule block; + private Item item; + private AtomicReference head; + private AtomicBoolean accepted; + + @Before + public void setUp() throws Exception { + adv = spy(new AdvService()); + delegate = mock(TronNetDelegate.class); + doCallRealMethod().when(delegate).recordInventory(any(), any(), anyLong()); + doCallRealMethod().when(delegate).confirmBlockInventory(any()); + inventory = new InventoryMsgHandler(); + blocks = new BlockMsgHandler(); + provider = PeerBlockTestSupport.peer(18888); + advertiser = PeerBlockTestSupport.peer(18889); + block = PeerBlockTestSupport.block(101); + item = new Item(block.getBlockId(), InventoryType.BLOCK); + head = new AtomicReference<>(block.getParentBlockId()); + accepted = new AtomicBoolean(); + provider.getAdvInvRequest().put(item, System.currentTimeMillis()); + when(delegate.getActivePeer()).thenReturn(Arrays.asList(provider, advertiser)); + when(delegate.getHeadBlockId()).thenAnswer(call -> head.get()); + when(delegate.containBlock(block.getParentBlockId())).thenReturn(true); + when(delegate.containBlock(block.getBlockId())).thenAnswer(call -> accepted.get()); + when(delegate.validBlock(block)).thenReturn(true); + doAnswer(call -> { + acceptBlock(); + return null; + }).when(delegate).processBlock(eq(block), anyBoolean()); + ReflectUtils.setFieldValue(adv, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(adv, "fastForward", false); + // Exercise inventory admission independently of provider selection and background workers. + doReturn(false).when(adv).addInv(any()); + ReflectUtils.setFieldValue(inventory, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(inventory, "advService", adv); + ReflectUtils.setFieldValue(inventory, "transactionsMsgHandler", + mock(TransactionsMsgHandler.class)); + ReflectUtils.setFieldValue(blocks, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(blocks, "advService", adv); + FetchBlockService fetch = mock(FetchBlockService.class); + doCallRealMethod().when(fetch).blockFetchSuccess(any(), any()); + ReflectUtils.setFieldValue(blocks, "fetchBlockService", fetch); + ReflectUtils.setFieldValue(blocks, "syncService", mock(SyncService.class)); + ReflectUtils.setFieldValue(blocks, "witnessProductBlockService", + mock(WitnessProductBlockService.class)); + ReflectUtils.setFieldValue(blocks, "fastForward", false); + } + + @After + public void tearDown() { + adv.close(); + } + + @Test + public void testAnnouncementWaitsForSuccessfulBlockProcessing() throws Exception { + record(100); + doAnswer(call -> { + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + Assert.assertNotNull(adv.getMessage(item)); + acceptBlock(); + return null; + }).when(delegate).processBlock(block, false); + + processBlock(); + + Assert.assertEquals(100, advertiser.getLastInteractiveTime()); + Assert.assertEquals(0, advertiser.getBlockRcvTime()); + Assert.assertTrue(provider.getBlockRcvTime() > 0); + Assert.assertNull(advertiser.getAdvBlockInvReceive().getIfPresent(item)); + } + + @Test + public void testInventoryIsRegisteredBeforeFetchCanComplete() throws Exception { + AtomicLong receivedAt = new AtomicLong(); + doAnswer(call -> { + Long pending = advertiser.getAdvBlockInvReceive().getIfPresent(item); + Assert.assertNotNull(pending); + receivedAt.set(pending); + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + processBlock(); + return false; + }).when(adv).addInv(item); + + inventory.processMessage(advertiser, inventory(item)); + + Assert.assertEquals(block.getBlockId(), head.get()); + Assert.assertEquals(receivedAt.get(), advertiser.getLastInteractiveTime()); + Assert.assertNull(advertiser.getAdvBlockInvReceive().getIfPresent(item)); + } + + @Test + public void testEqualAndLowerInventoryCannotRefreshActivity() throws Exception { + head.set(block.getBlockId()); + inventory.processMessage(advertiser, inventory(item)); + Item lower = new Item(block.getParentBlockId(), InventoryType.BLOCK); + inventory.processMessage(advertiser, inventory(lower)); + + delegate.confirmBlockInventory(block.getBlockId()); + delegate.confirmBlockInventory(block.getParentBlockId()); + + Assert.assertEquals(0, advertiser.getAdvBlockInvReceive().size()); + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + } + + @Test + public void testPreviouslySpreadInventoryCannotRefreshActivity() throws Exception { + advertiser.getAdvInvSpread().put(item, System.currentTimeMillis()); + inventory.processMessage(advertiser, inventory(item)); + + processBlock(); + + Assert.assertEquals(0, advertiser.getAdvBlockInvReceive().size()); + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + } + + @Test + public void testTransactionInventoryCannotRefreshBlockActivity() throws Exception { + inventory.processMessage(advertiser, inventory(new Item(item.getHash(), InventoryType.TRX))); + + processBlock(); + + Assert.assertEquals(0, advertiser.getAdvBlockInvReceive().size()); + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + } + + @Test + public void testSpreadingAfterReceiptDoesNotChangeEligibility() throws Exception { + record(100); + advertiser.getAdvInvSpread().put(item, 200L); + + processBlock(); + + Assert.assertEquals(100, advertiser.getLastInteractiveTime()); + } + + @Test + public void testLateDuplicateCannotReplaceEligibleReceiptTime() throws Exception { + record(100); + doAnswer(call -> { + acceptBlock(); + delegate.recordInventory(advertiser, item, 200); + Assert.assertEquals(Long.valueOf(200), advertiser.getAdvInvReceive().getIfPresent(item)); + Assert.assertEquals(Long.valueOf(100), advertiser.getAdvBlockInvReceive().getIfPresent(item)); + return null; + }).when(delegate).processBlock(block, false); + + processBlock(); + delegate.recordInventory(advertiser, item, 300); + delegate.confirmBlockInventory(block.getBlockId()); + + Assert.assertEquals(100, advertiser.getLastInteractiveTime()); + Assert.assertNull(advertiser.getAdvBlockInvReceive().getIfPresent(item)); + } + + @Test + public void testDifferentHashAtSameHeightCannotConfirmInventory() throws Exception { + record(100); + BlockId different = new BlockId(Sha256Hash.ZERO_HASH, block.getNum()); + Assert.assertNotEquals(block.getBlockId(), different); + + delegate.confirmBlockInventory(different); + + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + Assert.assertEquals(Long.valueOf(100), advertiser.getAdvBlockInvReceive().getIfPresent(item)); + processBlock(); + Assert.assertEquals(100, advertiser.getLastInteractiveTime()); + } + + @Test + public void testBadSignatureDoesNotConfirmInventory() throws Exception { + assertInvalidBlockDoesNotConfirm(TypeEnum.BLOCK_SIGN_INVALID); + } + + @Test + public void testBadMerkleDoesNotConfirmInventory() throws Exception { + assertInvalidBlockDoesNotConfirm(TypeEnum.BLOCK_MERKLE_INVALID); + } + + @Test + public void testInactiveWitnessDoesNotConfirmInventory() throws Exception { + record(100); + when(delegate.validBlock(block)).thenReturn(false); + + processBlock(); + + assertUnconfirmed(); + } + + @Test + public void testExecutionFailureDoesNotConfirmCachedInventory() throws Exception { + record(100); + doThrow(new P2pException(TypeEnum.BAD_BLOCK, "execution failed")) + .when(delegate).processBlock(block, false); + + processBlock(); + + Assert.assertNotNull(adv.getMessage(item)); + assertUnconfirmed(); + } + + @Test + public void testShutdownDoesNotConfirmInventory() throws Exception { + record(100); + when(delegate.isHitDown()).thenReturn(true); + + processBlock(); + + assertUnconfirmed(); + } + + @Test + public void testConfirmationDoesNotOverwriteNewerInteraction() throws Exception { + record(100); + advertiser.updateLastInteractiveTime(200); + + processBlock(); + + Assert.assertEquals(200, advertiser.getLastInteractiveTime()); + Assert.assertNull(advertiser.getAdvBlockInvReceive().getIfPresent(item)); + } + + @Test + public void testPendingInventoryEvictionDoesNotAffectBlockFetch() throws Exception { + record(100); + Long requestTime = provider.getAdvInvRequest().get(item); + for (int i = 1; i <= 200; i++) { + Item another = new Item(new BlockId(block.getBlockId(), block.getNum() + i), + InventoryType.BLOCK); + delegate.recordInventory(advertiser, another, 200); + } + Assert.assertTrue(advertiser.getAdvBlockInvReceive().size() > 0); + Assert.assertTrue(advertiser.getAdvBlockInvReceive().size() <= 100); + Assert.assertNull(advertiser.getAdvBlockInvReceive().getIfPresent(item)); + Assert.assertEquals(Long.valueOf(100), advertiser.getAdvInvReceive().getIfPresent(item)); + Assert.assertEquals(requestTime, provider.getAdvInvRequest().get(item)); + + processBlock(); + + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + Assert.assertEquals(0, advertiser.getBlockRcvTime()); + Assert.assertTrue(provider.getBlockRcvTime() > 0); + Assert.assertFalse(provider.getAdvInvRequest().containsKey(item)); + } + + @Test + public void testPendingInventoryExpiresAfterOneMinuteWithoutExtendingOnRead() throws Exception { + AtomicLong nanos = new AtomicLong(); + Ticker ticker = new Ticker() { + @Override + public long read() { + return nanos.get(); + } + }; + try (MockedStatic tickers = mockStatic(Ticker.class)) { + tickers.when(Ticker::systemTicker).thenReturn(ticker); + // Construct the production cache with a controlled clock, retaining its actual policy. + advertiser = PeerBlockTestSupport.peer(18890); + when(delegate.getActivePeer()).thenReturn(Arrays.asList(provider, advertiser)); + record(100); + Long requestTime = provider.getAdvInvRequest().get(item); + nanos.set(TimeUnit.MINUTES.toNanos(1) - 1); + Assert.assertEquals(Long.valueOf(100), advertiser.getAdvBlockInvReceive().getIfPresent(item)); + nanos.incrementAndGet(); + Assert.assertEquals(Long.valueOf(100), advertiser.getAdvInvReceive().getIfPresent(item)); + Assert.assertEquals(requestTime, provider.getAdvInvRequest().get(item)); + + processBlock(); + + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + Assert.assertEquals(0, advertiser.getBlockRcvTime()); + Assert.assertNull(advertiser.getAdvBlockInvReceive().getIfPresent(item)); + Assert.assertTrue(provider.getBlockRcvTime() > 0); + Assert.assertFalse(provider.getAdvInvRequest().containsKey(item)); + } + } + + @Test(timeout = 10_000) + public void testConcurrentConfirmationCannotMissEligibleInventory() throws Exception { + CountDownLatch headRead = new CountDownLatch(1); + CountDownLatch finishRegistration = new CountDownLatch(1); + CountDownLatch confirmationStarted = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + when(delegate.getHeadBlockId()).thenAnswer(call -> { + BlockId snapshot = head.get(); + headRead.countDown(); + Assert.assertTrue(finishRegistration.await(5, TimeUnit.SECONDS)); + return snapshot; + }); + try { + Future registration = executor.submit( + () -> delegate.recordInventory(advertiser, item, 100)); + Assert.assertTrue(headRead.await(5, TimeUnit.SECONDS)); + Assert.assertNull(advertiser.getAdvInvReceive().getIfPresent(item)); + acceptBlock(); + Future confirmation = executor.submit(() -> { + confirmationStarted.countDown(); + delegate.confirmBlockInventory(block.getBlockId()); + }); + Assert.assertTrue(confirmationStarted.await(5, TimeUnit.SECONDS)); + try { + confirmation.get(100, TimeUnit.MILLISECONDS); + Assert.fail("Confirmation must wait for the eligible inventory to be registered"); + } catch (TimeoutException expected) { + // The receipt observed the old head; it must be registered before confirmation finishes. + } + finishRegistration.countDown(); + registration.get(5, TimeUnit.SECONDS); + confirmation.get(5, TimeUnit.SECONDS); + + Assert.assertEquals(100, advertiser.getLastInteractiveTime()); + Assert.assertEquals(Long.valueOf(100), advertiser.getAdvInvReceive().getIfPresent(item)); + Assert.assertNull(advertiser.getAdvBlockInvReceive().getIfPresent(item)); + } finally { + finishRegistration.countDown(); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + private void record(long receivedAt) { + delegate.recordInventory(advertiser, item, receivedAt); + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + Assert.assertEquals(Long.valueOf(receivedAt), + advertiser.getAdvBlockInvReceive().getIfPresent(item)); + } + + private void acceptBlock() { + head.set(block.getBlockId()); + accepted.set(true); + } + + private void processBlock() throws Exception { + blocks.processMessage(provider, new BlockMessage(block)); + } + + private InventoryMessage inventory(Item value) { + return new InventoryMessage(Collections.singletonList(value.getHash()), value.getType()); + } + + private void assertInvalidBlockDoesNotConfirm(TypeEnum type) throws Exception { + record(100); + when(delegate.validBlock(block)).thenThrow(new P2pException(type, "invalid block")); + try { + processBlock(); + Assert.fail("Expected block validation to fail"); + } catch (P2pException e) { + Assert.assertEquals(type, e.getType()); + } + assertUnconfirmed(); + } + + private void assertUnconfirmed() { + Assert.assertEquals(1, advertiser.getLastInteractiveTime()); + Assert.assertEquals(0, advertiser.getBlockRcvTime()); + Assert.assertEquals(Long.valueOf(100), advertiser.getAdvBlockInvReceive().getIfPresent(item)); + } +} diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/BlockMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/BlockMsgHandlerTest.java index 48d7e93bc1f..6959c14823a 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/BlockMsgHandlerTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/BlockMsgHandlerTest.java @@ -190,6 +190,7 @@ public void testPaddedWitnessIngress() throws Exception { ByteString signature = canonical.getInstance().getBlockHeader().getWitnessSignature(); for (boolean strict : new boolean[]{false, true}) { TronNetDelegate delegate = witnessDelegate(strict, canonical.getWitnessAddress()); + Mockito.doReturn(false).when(delegate).containBlock(canonical.getBlockId()); AdvService adv = Mockito.mock(AdvService.class); BlockMsgHandler ingress = witnessHandler(delegate, adv, Mockito.mock(SyncService.class)); BlockMessage message = withWitnessSignature(canonical, @@ -303,7 +304,9 @@ private BlockMsgHandler witnessHandler(TronNetDelegate delegate, AdvService adv, ReflectUtils.setFieldValue(ingress, "tronNetDelegate", delegate); ReflectUtils.setFieldValue(ingress, "advService", adv); ReflectUtils.setFieldValue(ingress, "syncService", sync); - ReflectUtils.setFieldValue(ingress, "fetchBlockService", Mockito.mock(FetchBlockService.class)); + FetchBlockService fetch = Mockito.mock(FetchBlockService.class); + Mockito.doCallRealMethod().when(fetch).blockFetchSuccess(any(), any()); + ReflectUtils.setFieldValue(ingress, "fetchBlockService", fetch); ReflectUtils.setFieldValue(ingress, "witnessProductBlockService", Mockito.mock(WitnessProductBlockService.class)); ReflectUtils.setFieldValue(ingress, "fastForward", false); diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java index 0b2aad3f255..0f353614152 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java @@ -2,6 +2,7 @@ import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.Arrays; import java.util.Deque; @@ -25,6 +26,7 @@ import org.tron.core.net.message.sync.ChainInventoryMessage; import org.tron.core.net.peer.PeerConnection; import org.tron.core.net.service.sync.SyncService; +import org.tron.p2p.connection.Channel; public class ChainInventoryMsgHandlerTest { @@ -189,6 +191,39 @@ public void testSyncNextDecisionIsMadeUnderBlockLock() throws Exception { Assert.assertTrue(syncNextCalledUnderLock.get()); } + @Test + public void testPruningKnownBlockWakesPendingSyncBlockHandler() throws Exception { + Object blockLock = new Object(); + TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class); + SyncService syncService = Mockito.mock(SyncService.class); + BlockId known = new BlockId(Sha256Hash.ZERO_HASH, 1); + BlockId next = new BlockId(Sha256Hash.ZERO_HASH, 2); + AtomicBoolean handlerWokenUnderLock = new AtomicBoolean(); + + Mockito.when(tronNetDelegate.getBlockLock()).thenReturn(blockLock); + Mockito.when(tronNetDelegate.getHeadBlockId()) + .thenReturn(new BlockId(Sha256Hash.ZERO_HASH, 0)); + Mockito.when(tronNetDelegate.containBlock(known)).thenReturn(true); + Mockito.doAnswer(invocation -> { + handlerWokenUnderLock.set(Thread.holdsLock(blockLock)); + return null; + }).when(syncService).setHandleFlag(true); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", tronNetDelegate); + ReflectUtils.setFieldValue(handler, "syncService", syncService); + Channel channel = Mockito.mock(Channel.class); + Mockito.when(channel.getInetAddress()) + .thenReturn(new InetSocketAddress("127.0.0.2", 10001).getAddress()); + peer.setChannel(channel); + peer.setSyncChainRequested(new Pair<>(new LinkedList<>(Arrays.asList(known)), + System.currentTimeMillis())); + peer.getSyncBlockToFetch().add(known); + + handler.processMessage(peer, new ChainInventoryMessage(Arrays.asList(known, next), 0L)); + + Assert.assertEquals(next, peer.getSyncBlockToFetch().peek()); + Assert.assertTrue(handlerWokenUnderLock.get()); + } + @Test public void testNegativeRemainNumRejected() throws Exception { assertCheckRejects(createContinuousBlockIds(0L), -1L); diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java index 44e13ba8210..9ec8589597e 100644 --- a/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java +++ b/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java @@ -49,6 +49,7 @@ 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.sync.SyncService; import org.tron.p2p.connection.Channel; import org.tron.protos.Protocol; @@ -591,6 +592,7 @@ private void assertDisconnectRecovery(boolean announced, boolean busy) throws Ex List activePeers = new ArrayList<>(Collections.singletonList(source)); Mockito.when(delegate.getActivePeer()).thenReturn(activePeers); ReflectUtils.setFieldValue(adv, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(adv, "fetchBlockService", Mockito.mock(FetchBlockService.class)); setField(handler, "tronNetDelegate", delegate); setField(handler, "advService", adv); setField(handler, "chainBaseManager", Mockito.mock(ChainBaseManager.class)); diff --git a/framework/src/test/java/org/tron/core/net/peer/PeerBlockIdleTest.java b/framework/src/test/java/org/tron/core/net/peer/PeerBlockIdleTest.java new file mode 100644 index 00000000000..69d4aef4fe7 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/peer/PeerBlockIdleTest.java @@ -0,0 +1,42 @@ +package org.tron.core.net.peer; + +import java.util.ArrayDeque; +import org.junit.Assert; +import org.junit.Test; +import org.tron.common.utils.Pair; +import org.tron.common.utils.Sha256Hash; +import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.protos.Protocol.Inventory.InventoryType; + +public class PeerBlockIdleTest { + + @Test + public void testTransactionRequestDoesNotBlockBlockFetch() { + PeerConnection peer = new PeerConnection(); + Item trx = new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX); + peer.getAdvInvRequest().put(trx, System.currentTimeMillis()); + Assert.assertFalse(peer.isIdle()); + Assert.assertTrue(peer.isBlockFetchIdle()); + + Item block = new Item(new BlockId(), InventoryType.BLOCK); + peer.getAdvInvRequest().put(block, System.currentTimeMillis()); + Assert.assertFalse(peer.isBlockFetchIdle()); + peer.getAdvInvRequest().remove(block); + Assert.assertTrue(peer.isBlockFetchIdle()); + } + + @Test + public void testSyncRequestAndProcessingExcludeBlockProvider() { + PeerConnection peer = new PeerConnection(); + peer.getSyncBlockRequested().put(new BlockId(), System.currentTimeMillis()); + Assert.assertFalse(peer.isBlockFetchIdle()); + peer.getSyncBlockRequested().clear(); + peer.setSyncChainRequested(new Pair<>(new ArrayDeque<>(), System.currentTimeMillis())); + Assert.assertFalse(peer.isBlockFetchIdle()); + peer.setSyncChainRequested(null); + peer.getSyncBlockInProcess().add(new BlockId()); + Assert.assertFalse(peer.isBlockFetchIdle()); + peer.getSyncBlockInProcess().clear(); + Assert.assertTrue(peer.isBlockFetchIdle()); + } +} diff --git a/framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java b/framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java index 8d6baf73387..bdd9fcc083d 100644 --- a/framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java +++ b/framework/src/test/java/org/tron/core/net/peer/PeerConnectionTest.java @@ -86,6 +86,8 @@ public void testOnDisconnect() { Long time = System.currentTimeMillis(); BlockCapsule.BlockId blockId = new BlockCapsule.BlockId(); peerConnection.getAdvInvReceive().put(item, time); + peerConnection.getAdvBlockInvReceive().put( + new Item(blockId, Protocol.Inventory.InventoryType.BLOCK), time); peerConnection.getAdvInvSpread().put(item, time); peerConnection.getSyncBlockIdCache().put(item.getHash(), time); peerConnection.getSyncBlockToFetch().add(blockId); @@ -95,6 +97,7 @@ public void testOnDisconnect() { peerConnection.onDisconnect(); Assert.assertEquals(0, peerConnection.getAdvInvReceive().size()); + Assert.assertEquals(0, peerConnection.getAdvBlockInvReceive().size()); Assert.assertEquals(0, peerConnection.getAdvInvSpread().size()); Assert.assertEquals(0, peerConnection.getSyncBlockIdCache().size()); Assert.assertEquals(0, peerConnection.getSyncBlockToFetch().size()); @@ -102,6 +105,16 @@ public void testOnDisconnect() { Assert.assertEquals(0, peerConnection.getSyncBlockInProcess().size()); } + @Test + public void testLastInteractiveTimeOnlyAdvances() { + PeerConnection peer = new PeerConnection(); + peer.updateLastInteractiveTime(200); + peer.updateLastInteractiveTime(100); + Assert.assertEquals(200, peer.getLastInteractiveTime()); + peer.updateLastInteractiveTime(300); + Assert.assertEquals(300, peer.getLastInteractiveTime()); + } + @Test public void testIsIdle() { PeerConnection peerConnection = new PeerConnection(); diff --git a/framework/src/test/java/org/tron/core/net/service/fetchblock/FetchBlockRetryTest.java b/framework/src/test/java/org/tron/core/net/service/fetchblock/FetchBlockRetryTest.java new file mode 100644 index 00000000000..931ee169086 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/service/fetchblock/FetchBlockRetryTest.java @@ -0,0 +1,792 @@ +package org.tron.core.net.service.fetchblock; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.stubbing.Answer; +import org.tron.common.parameter.CommonParameter; +import org.tron.common.utils.ReflectUtils; +import org.tron.common.utils.Sha256Hash; +import org.tron.core.ChainBaseManager; +import org.tron.core.capsule.BlockCapsule; +import org.tron.core.capsule.BlockCapsule.BlockId; +import org.tron.core.config.Parameter.NetConstants; +import org.tron.core.exception.P2pException; +import org.tron.core.exception.P2pException.TypeEnum; +import org.tron.core.net.PeerBlockTestSupport; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.message.adv.BlockMessage; +import org.tron.core.net.message.adv.FetchInvDataMessage; +import org.tron.core.net.message.adv.InventoryMessage; +import org.tron.core.net.messagehandler.BlockMsgHandler; +import org.tron.core.net.messagehandler.InventoryMsgHandler; +import org.tron.core.net.peer.Item; +import org.tron.core.net.peer.PeerConnection; +import org.tron.core.net.peer.PeerStatusCheck; +import org.tron.core.net.service.adv.AdvService; +import org.tron.protos.Protocol.Inventory.InventoryType; +import org.tron.protos.Protocol.ReasonCode; + +public class FetchBlockRetryTest { + + private FetchBlockService fetch; + private AdvService adv; + private InventoryMsgHandler inventoryHandler; + private TronNetDelegate delegate; + private ChainBaseManager chain; + private PeerConnection first; + private PeerConnection second; + private List peers; + private BlockCapsule block; + private Item item; + private int previousTimeout; + + @Before + public void setUp() { + previousTimeout = CommonParameter.getInstance().fetchBlockTimeout; + CommonParameter.getInstance().fetchBlockTimeout = 1_000; + fetch = new FetchBlockService(); + adv = new AdvService(); + inventoryHandler = new InventoryMsgHandler(); + delegate = mock(TronNetDelegate.class); + doCallRealMethod().when(delegate).recordInventory(any(), any(), anyLong()); + doCallRealMethod().when(delegate).confirmBlockInventory(any()); + chain = mock(ChainBaseManager.class); + first = PeerBlockTestSupport.peer(18888); + second = PeerBlockTestSupport.peer(18889); + peers = new ArrayList<>(Arrays.asList(first, second)); + block = PeerBlockTestSupport.block(85636071); + item = new Item(block.getBlockId(), InventoryType.BLOCK); + when(delegate.getActivePeer()).thenReturn(peers); + when(delegate.getHeadBlockId()).thenReturn(block.getParentBlockId()); + when(chain.getHeadBlockNum()).thenReturn(block.getNum() - 1); + ReflectUtils.setFieldValue(fetch, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(fetch, "chainBaseManager", chain); + ReflectUtils.setFieldValue(adv, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(adv, "fetchBlockService", fetch); + ReflectUtils.setFieldValue(inventoryHandler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(inventoryHandler, "advService", adv); + } + + @After + public void tearDown() { + fetch.close(); + adv.close(); + CommonParameter.getInstance().fetchBlockTimeout = previousTimeout; + } + + @Test + public void testShortTimeoutWithoutAlternativeRetainsState() throws Exception { + beginAgedFetch(2_000); + Object state = state(); + + tick(state); + + Assert.assertSame(state, state()); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + verify(second, never()).sendMessage(any()); + } + + @Test + public void testLateInventoryBypassesCacheForProviderSwitch() throws Exception { + inventoryHandler.processMessage(first, inventory()); + Assert.assertNotNull(state()); + Assert.assertFalse(adv.addInv(item)); + ageState(2_000); + long originalRequest = first.getAdvInvRequest().get(item); + second.getAdvInvRequest().put(new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX), + System.currentTimeMillis()); + + inventoryHandler.processMessage(second, inventory()); + tick(state()); + + verify(second).sendMessage(any(FetchInvDataMessage.class)); + Assert.assertNotNull(state()); + Assert.assertSame(second, ReflectUtils.getFieldObject(state(), "peer")); + Assert.assertEquals(Long.valueOf(originalRequest), first.getAdvInvRequest().get(item)); + Assert.assertTrue(second.getAdvInvRequest().containsKey(item)); + Assert.assertEquals(1, first.getLastInteractiveTime()); + Assert.assertEquals(1, second.getLastInteractiveTime()); + Assert.assertEquals(0, second.getBlockRcvTime()); + } + + @Test + public void testFirstProviderCanHavePendingTransactions() throws Exception { + first.getAdvInvRequest().put(new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX), + System.currentTimeMillis()); + + inventoryHandler.processMessage(first, inventory()); + + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + verify(first).sendMessage(any(FetchInvDataMessage.class)); + } + + @Test + public void testInitialFetchPrefersIdleProviderOverPendingTransactions() { + Item transaction = new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX); + long now = System.currentTimeMillis(); + first.getAdvInvRequest().put(transaction, now); + first.getAdvInvReceive().put(item, now); + second.getAdvInvReceive().put(item, now); + + Assert.assertTrue(adv.addInv(item)); + + assertBlockRequests(second, item); + verify(first, never()).sendMessage(any()); + Assert.assertEquals(Long.valueOf(now), first.getAdvInvRequest().get(transaction)); + Assert.assertFalse(first.getAdvInvRequest().containsKey(item)); + Assert.assertTrue(second.getAdvInvRequest().containsKey(item)); + Assert.assertSame(second, ReflectUtils.getFieldObject(state(), "peer")); + } + + @Test + public void testInitialFetchPrefersFewerBatchRequestsOverLatency() { + Item transaction = new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX); + long now = System.currentTimeMillis(); + first.getAdvInvReceive().put(transaction, now); + first.getAdvInvReceive().put(item, now); + second.getAdvInvReceive().put(item, now); + Assert.assertTrue(adv.addInv(transaction)); + + setBlockLatencies(first, 10L, second, 500L); + Assert.assertTrue(adv.addInv(item)); + + assertBlockRequests(second, item); + Assert.assertTrue(first.getAdvInvRequest().containsKey(transaction)); + Assert.assertFalse(first.getAdvInvRequest().containsKey(item)); + Assert.assertTrue(second.getAdvInvRequest().containsKey(item)); + Assert.assertSame(second, ReflectUtils.getFieldObject(state(), "peer")); + } + + @Test + public void testInitialFetchDoesNotPreferPeerWithoutLatencySamples() { + long now = System.currentTimeMillis(); + first.getAdvInvReceive().put(item, now); + second.getAdvInvReceive().put(item, now); + + setBlockLatencies(first, 500L, second, null); + Assert.assertTrue(adv.addInv(item)); + + assertBlockRequests(first, item); + verify(second, never()).sendMessage(any()); + Assert.assertSame(first, ReflectUtils.getFieldObject(state(), "peer")); + } + + @Test + public void testBackupFetchStillPrefersLowerLatency() throws Exception { + PeerConnection faster = PeerBlockTestSupport.peer(18890); + peers.add(faster); + beginAgedFetch(2_000); + Long originalRequestTime = first.getAdvInvRequest().get(item); + long now = System.currentTimeMillis(); + second.getAdvInvReceive().put(item, now); + faster.getAdvInvReceive().put(item, now); + Item transaction = new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX); + faster.getAdvInvRequest().put(transaction, now); + + setBlockLatencies(second, 500L, faster, 10L); + tick(state()); + + assertBlockRequests(faster, item); + verify(second, never()).sendMessage(any()); + Assert.assertTrue(faster.getAdvInvRequest().containsKey(transaction)); + Assert.assertTrue(faster.getAdvInvRequest().containsKey(item)); + Assert.assertEquals(originalRequestTime, first.getAdvInvRequest().get(item)); + Assert.assertSame(faster, ReflectUtils.getFieldObject(state(), "peer")); + } + + @Test + public void testBackupRequestsAreLimitedPerBlock() throws Exception { + for (int i = 0; i < 4; i++) { + peers.add(PeerBlockTestSupport.peer(18890 + i)); + } + peers.forEach(peer -> peer.getAdvInvReceive().put(item, System.currentTimeMillis())); + beginAgedFetch(2_000); + Long originalRequestTime = first.getAdvInvRequest().get(item); + + setBlockLatencies(peers, 2_000L, 2_000L, 2_000L, 2_000L, 2_000L, 2_000L); + tick(state()); + Assert.assertSame(second, ReflectUtils.getFieldObject(state(), "peer")); + ageState(2_000); + Object backupState = state(); + Long backupRequestTime = second.getAdvInvRequest().get(item); + for (int i = 0; i < 10; i++) { + // An eligible third provider and an expired short timeout must not bypass the cap. + Assert.assertTrue(fetch.canFetchBlock(peers.get(2), item, System.currentTimeMillis())); + tick(state()); + } + Assert.assertSame(backupState, state()); + Assert.assertEquals(backupRequestTime, second.getAdvInvRequest().get(item)); + + assertBlockRequests(second, item); + verify(first, never()).sendMessage(any()); + for (int i = 2; i < peers.size(); i++) { + verify(peers.get(i), never()).sendMessage(any()); + } + Assert.assertEquals(2L, peers.stream() + .filter(peer -> peer.getAdvInvRequest().containsKey(item)).count()); + Assert.assertEquals(originalRequestTime, first.getAdvInvRequest().get(item)); + } + + @Test + public void testSlowBackupWaitsForActualTimeout() throws Exception { + // Leave enough time for assertions without relying on sleeps or a sub-second test run. + ReflectUtils.setFieldValue(fetch, "fetchTimeOut", 10_000L); + CommonParameter.getInstance().fetchBlockTimeout = 10_000; + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + + setBlockLatencies(first, 15_000L, second, 15_000L); + beginAgedFetch(100); + Object originalState = state(); + tick(state()); + + Assert.assertSame(originalState, state()); + Assert.assertFalse(second.getAdvInvRequest().containsKey(item)); + verify(second, never()).sendMessage(any()); + + ageState(11_000); + Long originalRequestTime = first.getAdvInvRequest().get(item); + tick(state()); + + assertBlockRequests(second, item); + Assert.assertSame(second, ReflectUtils.getFieldObject(state(), "peer")); + Assert.assertEquals(originalRequestTime, first.getAdvInvRequest().get(item)); + } + + @Test + public void testSaturatedLatencyDoesNotDispatchBackupWithNoTimeBenefit() throws Exception { + assertSaturatedOldPeerWaitsForBenefit(9_950L); + } + + @Test + public void testSaturatedLatencyRequiresSufficientSpeedup() throws Exception { + assertSaturatedOldPeerWaitsForBenefit(6_000L); + } + + @Test + public void testFasterBackupCanStartBeforeTimeout() throws Exception { + ReflectUtils.setFieldValue(fetch, "fetchTimeOut", 10_000L); + CommonParameter.getInstance().fetchBlockTimeout = 10_000; + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + + setBlockLatencies(first, 8_000L, second, 100L); + beginAgedFetch(100); + Long originalRequestTime = first.getAdvInvRequest().get(item); + tick(state()); + + assertBlockRequests(second, item); + Assert.assertTrue(System.currentTimeMillis() - originalRequestTime < 10_000); + Assert.assertEquals(originalRequestTime, first.getAdvInvRequest().get(item)); + Assert.assertSame(second, ReflectUtils.getFieldObject(state(), "peer")); + } + + @Test + public void testOtherRequestsDoNotUseBlockBackupLimit() throws Exception { + PeerConnection otherProvider = PeerBlockTestSupport.peer(18890); + peers.add(otherProvider); + Item otherBlock = new Item(block.getParentBlockId(), InventoryType.BLOCK); + Item transaction = new Item(item.getHash(), InventoryType.TRX); + long now = System.currentTimeMillis(); + otherProvider.getAdvInvRequest().put(otherBlock, now); + second.getAdvInvRequest().put(transaction, now); + second.getAdvInvReceive().put(item, now); + beginAgedFetch(2_000); + + tick(state()); + + assertBlockRequests(second, item); + Assert.assertTrue(second.getAdvInvRequest().containsKey(item)); + Assert.assertEquals(Long.valueOf(now), second.getAdvInvRequest().get(transaction)); + Assert.assertEquals(Long.valueOf(now), otherProvider.getAdvInvRequest().get(otherBlock)); + verify(otherProvider, never()).sendMessage(any()); + } + + @Test + public void testOriginalDisconnectAllowsReplacementBackup() throws Exception { + PeerConnection replacement = PeerBlockTestSupport.peer(18890); + peers.add(replacement); + beginAgedFetch(2_000); + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + tick(state()); + ageState(2_000); + Long backupRequestTime = second.getAdvInvRequest().get(item); + replacement.getAdvInvReceive().put(item, System.currentTimeMillis()); + tick(state()); + verify(replacement, never()).sendMessage(any()); + + disconnect(first); + tick(state()); + + assertBlockRequests(second, item); + assertBlockRequests(replacement, item); + Assert.assertEquals(backupRequestTime, second.getAdvInvRequest().get(item)); + Assert.assertTrue(replacement.getAdvInvRequest().containsKey(item)); + Assert.assertSame(replacement, ReflectUtils.getFieldObject(state(), "peer")); + } + + @Test + public void testBackupDisconnectAllowsReplacementBackup() throws Exception { + PeerConnection replacement = PeerBlockTestSupport.peer(18890); + peers.add(replacement); + beginAgedFetch(2_000); + Long originalRequestTime = first.getAdvInvRequest().get(item); + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + tick(state()); + ageState(2_000); + replacement.getAdvInvReceive().put(item, System.currentTimeMillis()); + tick(state()); + verify(replacement, never()).sendMessage(any()); + + disconnect(second); + Assert.assertSame(first, ReflectUtils.getFieldObject(state(), "peer")); + Assert.assertEquals(originalRequestTime, ReflectUtils.getFieldObject(state(), "time")); + tick(state()); + + assertBlockRequests(second, item); + assertBlockRequests(replacement, item); + verify(first, never()).sendMessage(any()); + Assert.assertEquals(originalRequestTime, first.getAdvInvRequest().get(item)); + Assert.assertTrue(replacement.getAdvInvRequest().containsKey(item)); + Assert.assertSame(replacement, ReflectUtils.getFieldObject(state(), "peer")); + } + + @Test + public void testQueuedBlocksAreFetchedInHeightOrder() throws Exception { + List blocks = queueBlocksInReverseHeightOrder(); + Item transaction = new Item(Sha256Hash.ZERO_HASH, InventoryType.TRX); + first.getAdvInvRequest().put(transaction, System.currentTimeMillis()); + + consumeInventory(); + + assertBlockRequests(first, blocks.get(0)); + Assert.assertTrue(first.getAdvInvRequest().containsKey(blocks.get(0))); + Assert.assertTrue(first.getAdvInvRequest().containsKey(transaction)); + Assert.assertFalse(first.getAdvInvRequest().containsKey(blocks.get(1))); + Assert.assertFalse(pendingInventory().containsKey(blocks.get(0))); + Assert.assertTrue(pendingInventory().containsKey(blocks.get(1))); + Assert.assertEquals(blocks.get(0).getHash(), ReflectUtils.getFieldObject(state(), "hash")); + } + + @Test + public void testQueuedNextBlockFetchedAfterHeadAdvances() throws Exception { + List blocks = queueBlocksInReverseHeightOrder(); + consumeInventory(); + Assert.assertTrue(first.getAdvInvRequest().containsKey(blocks.get(0))); + Assert.assertFalse(first.getAdvInvRequest().containsKey(blocks.get(1))); + + fetch.blockFetchSuccess(blocks.get(0).getHash(), first); + when(chain.getHeadBlockNum()).thenReturn(block.getNum()); + when(delegate.getHeadBlockId()).thenReturn(new BlockId(blocks.get(0).getHash())); + + consumeInventory(); + + assertBlockRequests(first, blocks.get(0), blocks.get(1)); + Assert.assertFalse(first.getAdvInvRequest().containsKey(blocks.get(0))); + Assert.assertTrue(first.getAdvInvRequest().containsKey(blocks.get(1))); + Assert.assertTrue(pendingInventory().isEmpty()); + Assert.assertEquals(blocks.get(1).getHash(), ReflectUtils.getFieldObject(state(), "hash")); + } + + @Test + public void testOnlyHigherBlockCanStillBeFetched() throws Exception { + Item higher = new Item(new BlockId(Sha256Hash.ZERO_HASH, block.getNum() + 1), + InventoryType.BLOCK); + + inventoryHandler.processMessage(first, + new InventoryMessage(Collections.singletonList(higher.getHash()), InventoryType.BLOCK)); + + assertBlockRequests(first, higher); + Assert.assertTrue(first.getAdvInvRequest().containsKey(higher)); + Assert.assertTrue(pendingInventory().isEmpty()); + } + + @Test + public void testBlockEligibilityExcludesBusySyncingAndDisconnectedPeers() { + first.getAdvInvReceive().put(item, System.currentTimeMillis()); + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + first.getSyncBlockInProcess().add(block.getParentBlockId()); + Assert.assertFalse(fetch.canFetchBlock(first, item, System.currentTimeMillis())); + Assert.assertTrue(fetch.canFetchBlock(second, item, System.currentTimeMillis())); + second.setNeedSyncFromPeer(true); + Assert.assertFalse(fetch.canFetchBlock(second, item, System.currentTimeMillis())); + second.setNeedSyncFromPeer(false); + second.setNeedSyncFromUs(true); + Assert.assertFalse(fetch.canFetchBlock(second, item, System.currentTimeMillis())); + second.setNeedSyncFromUs(false); + when(second.getChannel().isDisconnect()).thenReturn(true); + Assert.assertFalse(fetch.canFetchBlock(second, item, System.currentTimeMillis())); + } + + @Test + public void testBlockEligibilityRequiresRecentInventory() { + Assert.assertFalse(fetch.canFetchBlock(first, item, System.currentTimeMillis())); + first.getAdvInvReceive().put(item, + System.currentTimeMillis() - NetConstants.ADV_TIME_OUT - 1_000); + Assert.assertFalse(fetch.canFetchBlock(first, item, System.currentTimeMillis())); + } + + @Test + public void testOnlyOriginalProviderGetsFinalTimeout() throws Exception { + beginAgedFetch(2_000); + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + tick(state()); + long connectedAt = System.currentTimeMillis(); + when(first.getChannel().getStartTime()).thenReturn(connectedAt); + when(second.getChannel().getStartTime()).thenReturn(connectedAt); + first.getAdvInvRequest().put(item, + System.currentTimeMillis() - NetConstants.ADV_TIME_OUT - 1_000); + PeerStatusCheck status = new PeerStatusCheck(); + ReflectUtils.setFieldValue(status, "tronNetDelegate", delegate); + try { + status.statusCheck(); + } finally { + status.close(); + } + + verify(first).disconnect(ReasonCode.TIME_OUT); + verify(second, never()).disconnect(any()); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + Assert.assertTrue(second.getAdvInvRequest().containsKey(item)); + } + + @Test + public void testFinalDeadlineDoesNotDiscardResponsibility() throws Exception { + beginAgedFetch(NetConstants.ADV_TIME_OUT + 1_000); + Object state = state(); + + tick(state); + + Assert.assertSame(state, state()); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + verify(first, never()).disconnect(any()); + } + + @Test + public void testDisconnectInvalidatesCacheWhenNoAlternativeExists() throws Exception { + inventoryHandler.processMessage(first, inventory()); + disconnect(first); + Assert.assertNull(state()); + + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + Assert.assertTrue(adv.addInv(item)); + verify(second).sendMessage(any(FetchInvDataMessage.class)); + } + + @Test + public void testInvalidBlockDisconnectRetriesFromAlternative() throws Exception { + inventoryHandler.processMessage(first, inventory()); + inventoryHandler.processMessage(second, inventory()); + BlockMsgHandler handler = new BlockMsgHandler(); + ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(handler, "fetchBlockService", fetch); + ReflectUtils.setFieldValue(handler, "fastForward", false); + when(delegate.validBlock(block)) + .thenThrow(new P2pException(TypeEnum.BLOCK_MERKLE_INVALID, "bad")); + try { + handler.processMessage(first, new BlockMessage(block)); + Assert.fail("Expected a bad Merkle root to be rejected"); + } catch (P2pException e) { + Assert.assertEquals(TypeEnum.BLOCK_MERKLE_INVALID, e.getType()); + } + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + + disconnect(first); + + Assert.assertTrue(second.getAdvInvRequest().containsKey(item)); + verify(second).sendMessage(any(FetchInvDataMessage.class)); + Assert.assertNotNull(state()); + } + + @Test + public void testOriginalDisconnectDoesNotDuplicateOutstandingBackupRequest() throws Exception { + beginAgedFetch(2_000); + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + tick(state()); + Object backupState = state(); + + disconnect(first); + + Assert.assertSame(backupState, state()); + verify(second, times(1)).sendMessage(any(FetchInvDataMessage.class)); + } + + @Test + public void testBackupDisconnectRestoresTrackingOfOriginalRequest() throws Exception { + beginAgedFetch(2_000); + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + tick(state()); + + disconnect(second); + + Assert.assertNotNull(state()); + Assert.assertSame(first, ReflectUtils.getFieldObject(state(), "peer")); + verify(first, never()).sendMessage(any()); + } + + @Test(timeout = 10_000) + public void testRecoveryWaitsForCompletionRequestRemoval() throws Exception { + assertCompletionAndRecoveryAreSerialized(true); + } + + @Test(timeout = 10_000) + public void testCompletionWaitsForRecoveryRequestLookup() throws Exception { + assertCompletionAndRecoveryAreSerialized(false); + } + + @Test + public void testStaleWorkerCannotOverwriteNewFetch() throws Exception { + beginAgedFetch(2_000); + Object oldState = state(); + fetch.blockFetchSuccess(item.getHash(), first); + Item next = new Item(PeerBlockTestSupport.block(block.getNum() + 1).getBlockId(), + InventoryType.BLOCK); + when(chain.getHeadBlockNum()).thenReturn(block.getNum()); + second.getAdvInvRequest().put(next, System.currentTimeMillis()); + fetch.fetchBlock(Collections.singletonList(next.getHash()), second); + Object newState = state(); + + tick(oldState); + fetch.blockFetchSuccess(item.getHash(), first); + + Assert.assertSame(newState, state()); + verify(second, never()).sendMessage(any()); + } + + @Test + public void testImmediateFirstResponseDoesNotResurrectCompletedFetch() throws Exception { + doAnswer(call -> { + Assert.assertNotNull(state()); + fetch.blockFetchSuccess(item.getHash(), first); + return null; + }).when(first).sendMessage(any(FetchInvDataMessage.class)); + + inventoryHandler.processMessage(first, inventory()); + + Assert.assertNull(state()); + } + + @Test + public void testImmediateBackupResponseDoesNotResurrectCompletedFetch() throws Exception { + beginAgedFetch(2_000); + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + doAnswer(call -> { + fetch.blockFetchSuccess(item.getHash(), second); + return null; + }).when(second).sendMessage(any(FetchInvDataMessage.class)); + + tick(state()); + + Assert.assertNull(state()); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + } + + private InventoryMessage inventory() { + return new InventoryMessage(Collections.singletonList(item.getHash()), InventoryType.BLOCK); + } + + @Test + public void testNewHeadReplacesObsoleteFetchWithoutLosingOriginalRequest() { + beginAgedFetch(2_000); + Item next = new Item(PeerBlockTestSupport.block(block.getNum() + 1).getBlockId(), + InventoryType.BLOCK); + when(chain.getHeadBlockNum()).thenReturn(block.getNum()); + second.getAdvInvRequest().put(next, System.currentTimeMillis()); + + fetch.fetchBlock(Collections.singletonList(next.getHash()), second); + + Assert.assertEquals(next.getHash(), ReflectUtils.getFieldObject(state(), "hash")); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + } + + @Test + public void testWorkerStopsObsoleteFetchAfterHeadAdvances() throws Exception { + beginAgedFetch(2_000); + when(chain.getHeadBlockNum()).thenReturn(block.getNum()); + + tick(state()); + + Assert.assertNull(state()); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + } + + private void assertCompletionAndRecoveryAreSerialized(boolean completeFirst) throws Exception { + beginAgedFetch(2_000); + second.getAdvInvRequest().put(item, first.getAdvInvRequest().get(item)); + fetch.onDisconnect(first); + Map requests = spy(second.getAdvInvRequest()); + second.setAdvInvRequest(requests); + CountDownLatch requestAccessed = new CountDownLatch(1); + CountDownLatch finishAccess = new CountDownLatch(1); + CountDownLatch otherStarted = new CountDownLatch(1); + Answer pauseRequestAccess = invocation -> { + Long time = (Long) invocation.callRealMethod(); + requestAccessed.countDown(); + Assert.assertTrue(finishAccess.await(5, TimeUnit.SECONDS)); + return time; + }; + if (completeFirst) { + doAnswer(pauseRequestAccess).when(requests).remove(item); + } else { + doAnswer(pauseRequestAccess).when(requests).get(item); + } + Runnable complete = () -> fetch.blockFetchSuccess(item.getHash(), second); + // Recovery may already have selected this provider before the response completes. + Runnable recover = () -> fetch.fetchBlock(Collections.singletonList(item.getHash()), second); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future initial = executor.submit(completeFirst ? complete : recover); + Assert.assertTrue(requestAccessed.await(5, TimeUnit.SECONDS)); + Future other = executor.submit(() -> { + otherStarted.countDown(); + (completeFirst ? recover : complete).run(); + }); + Assert.assertTrue(otherStarted.await(5, TimeUnit.SECONDS)); + try { + other.get(100, TimeUnit.MILLISECONDS); + Assert.fail("Completion and recovery must not interleave their request updates"); + } catch (TimeoutException expected) { + // The competing operation must wait until the request access finishes. + } + finishAccess.countDown(); + initial.get(5, TimeUnit.SECONDS); + other.get(5, TimeUnit.SECONDS); + + Assert.assertNull(state()); + Assert.assertFalse(second.getAdvInvRequest().containsKey(item)); + Assert.assertTrue(first.getAdvInvRequest().containsKey(item)); + } finally { + finishAccess.countDown(); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + private void beginAgedFetch(long age) { + first.getAdvInvRequest().put(item, System.currentTimeMillis() - age); + fetch.fetchBlock(Collections.singletonList(item.getHash()), first); + Assert.assertNotNull(state()); + } + + private List queueBlocksInReverseHeightOrder() throws Exception { + peers.remove(second); + byte[] hash = new byte[Sha256Hash.LENGTH]; + hash[hash.length - 1] = 1; + Item lower = new Item(new BlockId(hash, block.getNum()), InventoryType.BLOCK); + Item higher = new Item(new BlockId(Sha256Hash.ZERO_HASH, block.getNum() + 1), + InventoryType.BLOCK); + Item busy = new Item(block.getParentBlockId(), InventoryType.BLOCK); + first.getAdvInvRequest().put(busy, System.currentTimeMillis()); + + inventoryHandler.processMessage(first, + new InventoryMessage(Arrays.asList(lower.getHash(), higher.getHash()), + InventoryType.BLOCK)); + + verify(first, never()).sendMessage(any(FetchInvDataMessage.class)); + Assert.assertEquals(2, pendingInventory().size()); + // Assert the fixture order so the test cannot pass merely because the map visits lower first. + Assert.assertEquals(Arrays.asList(higher, lower), + new ArrayList<>(pendingInventory().keySet())); + first.getAdvInvRequest().remove(busy); + return Arrays.asList(lower, higher); + } + + private Map pendingInventory() { + return (Map) ReflectUtils.getFieldObject(adv, "invToFetch"); + } + + private void consumeInventory() throws Exception { + Method method = AdvService.class.getDeclaredMethod("consumerInvToFetch"); + method.setAccessible(true); + method.invoke(adv); + } + + private void assertBlockRequests(PeerConnection peer, Item... blocks) { + ArgumentCaptor requests = + ArgumentCaptor.forClass(FetchInvDataMessage.class); + verify(peer, times(blocks.length)).sendMessage(requests.capture()); + for (int i = 0; i < blocks.length; i++) { + Assert.assertEquals(InventoryType.BLOCK, requests.getAllValues().get(i).getInventoryType()); + Assert.assertEquals(Collections.singletonList(blocks[i].getHash()), + requests.getAllValues().get(i).getHashList()); + } + } + + private void setBlockLatencies(PeerConnection left, Long leftLatency, + PeerConnection right, Long rightLatency) { + setBlockLatencies(Arrays.asList(left, right), leftLatency, rightLatency); + } + + private void setBlockLatencies(List candidates, + Long... latencies) { + Assert.assertEquals(candidates.size(), latencies.length); + for (int i = 0; i < candidates.size(); i++) { + if (latencies[i] != null) { + candidates.get(i).updateFetchLatency(latencies[i]); + } + } + } + + private void assertSaturatedOldPeerWaitsForBenefit(long candidateLatency) throws Exception { + ReflectUtils.setFieldValue(fetch, "fetchTimeOut", 10_000L); + CommonParameter.getInstance().fetchBlockTimeout = 10_000; + second.getAdvInvReceive().put(item, System.currentTimeMillis()); + setBlockLatencies(first, 10_000L, second, candidateLatency); + beginAgedFetch(100); + Object originalState = state(); + + tick(originalState); + + Assert.assertSame(originalState, state()); + Assert.assertFalse(second.getAdvInvRequest().containsKey(item)); + verify(second, never()).sendMessage(any()); + } + + private void ageState(long age) { + long time = System.currentTimeMillis() - age; + ReflectUtils.setFieldValue(state(), "time", time); + PeerConnection provider = (PeerConnection) ReflectUtils.getFieldObject(state(), "peer"); + provider.getAdvInvRequest().put(item, time); + } + + private Object state() { + return ReflectUtils.getFieldObject(fetch, "fetchBlockInfo"); + } + + private void tick(Object state) throws Exception { + Class stateClass = Class.forName(FetchBlockService.class.getName() + "$FetchBlockInfo"); + Method method = FetchBlockService.class.getDeclaredMethod("fetchBlockProcess", stateClass); + method.setAccessible(true); + method.invoke(fetch, state); + } + + private void disconnect(PeerConnection peer) { + when(peer.getChannel().isDisconnect()).thenReturn(true); + peers.remove(peer); + adv.onDisconnect(peer); + } +} diff --git a/framework/src/test/java/org/tron/core/net/service/sync/SyncBlockContributionTest.java b/framework/src/test/java/org/tron/core/net/service/sync/SyncBlockContributionTest.java new file mode 100644 index 00000000000..2de71ff9da8 --- /dev/null +++ b/framework/src/test/java/org/tron/core/net/service/sync/SyncBlockContributionTest.java @@ -0,0 +1,140 @@ +package org.tron.core.net.service.sync; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.doThrow; +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 java.lang.reflect.Method; +import java.util.Arrays; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.tron.common.utils.ReflectUtils; +import org.tron.core.capsule.BlockCapsule; +import org.tron.core.exception.P2pException; +import org.tron.core.exception.P2pException.TypeEnum; +import org.tron.core.net.PeerBlockTestSupport; +import org.tron.core.net.TronNetDelegate; +import org.tron.core.net.messagehandler.PbftDataSyncHandler; +import org.tron.core.net.peer.Item; +import org.tron.core.net.peer.PeerConnection; +import org.tron.protos.Protocol.Inventory.InventoryType; +import org.tron.protos.Protocol.ReasonCode; + +public class SyncBlockContributionTest { + + private SyncService sync; + private TronNetDelegate delegate; + private PeerConnection provider; + private PeerConnection other; + private BlockCapsule block; + + @Before + public void setUp() { + sync = new SyncService(); + delegate = mock(TronNetDelegate.class); + doCallRealMethod().when(delegate).recordInventory(any(), any(), anyLong()); + doCallRealMethod().when(delegate).confirmBlockInventory(any()); + provider = PeerBlockTestSupport.peer(18888); + other = PeerBlockTestSupport.peer(18889); + block = PeerBlockTestSupport.block(10); + provider.getSyncBlockToFetch().add(block.getBlockId()); + other.getSyncBlockToFetch().add(block.getBlockId()); + when(delegate.getActivePeer()).thenReturn(Arrays.asList(provider, other)); + when(delegate.getHeadBlockId()).thenReturn(block.getParentBlockId()); + ReflectUtils.setFieldValue(sync, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(sync, "pbftDataSyncHandler", mock(PbftDataSyncHandler.class)); + } + + @After + public void tearDown() { + sync.close(); + } + + @Test + public void testOnlyValidatedProviderGetsTimestamps() throws Exception { + process(); + Assert.assertTrue(provider.getLastInteractiveTime() > 1); + Assert.assertTrue(provider.getBlockRcvTime() > 0); + Assert.assertEquals(1, other.getLastInteractiveTime()); + Assert.assertEquals(0, other.getBlockRcvTime()); + } + + @Test + public void testInvalidSignatureOnlyBlamesProvider() throws Exception { + delegate.recordInventory(other, new Item(block.getBlockId(), InventoryType.BLOCK), 100); + doThrow(new P2pException(TypeEnum.BLOCK_SIGN_INVALID, "bad signature")) + .when(delegate).validSignature(block); + process(); + verify(provider).disconnect(ReasonCode.BAD_BLOCK); + verify(other, never()).disconnect(any()); + Assert.assertEquals(1, provider.getLastInteractiveTime()); + Assert.assertEquals(0, provider.getBlockRcvTime()); + Assert.assertEquals(1, other.getLastInteractiveTime()); + } + + @Test + public void testStateFailureUsesBadBlockWithoutImprovingTimestamps() throws Exception { + delegate.recordInventory(other, new Item(block.getBlockId(), InventoryType.BLOCK), 100); + doThrow(new P2pException(TypeEnum.BAD_BLOCK, "state failure")) + .when(delegate).processBlock(block, true); + process(); + verify(provider).disconnect(ReasonCode.BAD_BLOCK); + verify(other).disconnect(ReasonCode.BAD_BLOCK); + verify(provider, never()).disconnect(ReasonCode.SYNC_FAIL); + verify(other, never()).disconnect(ReasonCode.SYNC_FAIL); + Assert.assertEquals(1, provider.getLastInteractiveTime()); + Assert.assertEquals(0, provider.getBlockRcvTime()); + Assert.assertEquals(1, other.getLastInteractiveTime()); + } + + @Test + public void testShutdownDoesNotImproveTimestamps() throws Exception { + delegate.recordInventory(other, new Item(block.getBlockId(), InventoryType.BLOCK), 100); + when(delegate.isHitDown()).thenReturn(true); + process(); + Assert.assertEquals(1, provider.getLastInteractiveTime()); + Assert.assertEquals(0, provider.getBlockRcvTime()); + Assert.assertEquals(1, other.getLastInteractiveTime()); + } + + @Test + public void testSyncBlockConfirmsEligibleInventoryWithoutContribution() throws Exception { + Item item = new Item(block.getBlockId(), InventoryType.BLOCK); + delegate.recordInventory(other, item, 100); + Assert.assertEquals(1, other.getLastInteractiveTime()); + + process(); + + Assert.assertEquals(100, other.getLastInteractiveTime()); + Assert.assertEquals(0, other.getBlockRcvTime()); + Assert.assertNull(other.getAdvBlockInvReceive().getIfPresent(item)); + } + + @Test + public void testOldSyncBlockDoesNotGetContribution() throws Exception { + when(delegate.getHeadBlockId()).thenReturn(PeerBlockTestSupport.block(11).getBlockId()); + process(); + Assert.assertEquals(0, provider.getBlockRcvTime()); + } + + @Test + public void testKnownSyncBlockDoesNotGetContribution() throws Exception { + when(delegate.containBlock(block.getBlockId())).thenReturn(true); + process(); + Assert.assertEquals(0, provider.getBlockRcvTime()); + } + + private void process() throws Exception { + Method method = SyncService.class.getDeclaredMethod("processSyncBlock", + BlockCapsule.class, PeerConnection.class); + method.setAccessible(true); + method.invoke(sync, block, provider); + } +} diff --git a/framework/src/test/java/org/tron/core/net/services/AdvServiceTest.java b/framework/src/test/java/org/tron/core/net/services/AdvServiceTest.java index 808c5eb09e0..5c0ac553c15 100644 --- a/framework/src/test/java/org/tron/core/net/services/AdvServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/AdvServiceTest.java @@ -36,6 +36,7 @@ import org.tron.core.net.peer.PeerConnection; import org.tron.core.net.peer.PeerManager; import org.tron.core.net.service.adv.AdvService; +import org.tron.core.net.service.fetchblock.FetchBlockService; import org.tron.p2p.connection.Channel; import org.tron.protos.Protocol; import org.tron.protos.Protocol.Inventory.InventoryType; @@ -185,6 +186,7 @@ public void sendMessage(Message message) { Mockito.when(delegate.getActivePeer()).thenReturn(Arrays.asList(first, second)); AdvService isolated = new AdvService(); ReflectUtils.setFieldValue(isolated, "tronNetDelegate", delegate); + ReflectUtils.setFieldValue(isolated, "fetchBlockService", mock(FetchBlockService.class)); try { BlockCapsule block = new BlockCapsule(1, Sha256Hash.ZERO_HASH, System.currentTimeMillis(), Sha256Hash.ZERO_HASH.getByteString()); 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 index b7a6ecdfcdc..8de74c58532 100644 --- a/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/FetchBlockServiceTest.java @@ -7,7 +7,6 @@ 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; @@ -19,6 +18,8 @@ import org.tron.common.BaseMethodTest; import org.tron.common.utils.ReflectUtils; import org.tron.common.utils.Sha256Hash; +import org.tron.core.ChainBaseManager; +import org.tron.core.capsule.BlockCapsule.BlockId; import org.tron.core.net.TronNetDelegate; import org.tron.core.net.peer.Item; import org.tron.core.net.peer.PeerConnection; @@ -36,6 +37,7 @@ protected void afterInit() { service = context.getBean(FetchBlockService.class); tronNetDelegate = mock(TronNetDelegate.class); ReflectUtils.setFieldValue(service, "tronNetDelegate", tronNetDelegate); + ReflectUtils.setFieldValue(service, "chainBaseManager", mock(ChainBaseManager.class)); } /** @@ -67,9 +69,10 @@ public void testSelectLowestLatencyPeer() throws Exception { oldPeer.updateFetchLatency(200L); PeerConnection newPeer = context.getBean(PeerConnection.class); newPeer.setChannel(newChannel); + allowBlockFetch(newPeer); newPeer.updateFetchLatency(50L); - Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Sha256Hash hash = new BlockId(Sha256Hash.ZERO_HASH, 1); Item item = new Item(hash, InventoryType.BLOCK); // both peers advertise having the block @@ -101,15 +104,13 @@ public void testSelectLowestLatencyPeer() throws Exception { 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")); + // Keep tracking the backup so a third request cannot amplify this fetch. + assertTrackedPeer(newPeer); } - /** - * A 600ms seed is clamped to 500ms, and >= timeout triggers fast-switch. - */ + /** A much faster peer can still be selected before the actual timeout. */ @Test - public void testSwitchOnOldPeerTimeout() throws Exception { + public void testSwitchToMuchFasterPeerBeforeTimeout() throws Exception { InetSocketAddress oldAddr = new InetSocketAddress("127.0.0.3", 10001); InetSocketAddress newAddr = new InetSocketAddress("127.0.0.4", 10001); @@ -132,9 +133,10 @@ public void testSwitchOnOldPeerTimeout() throws Exception { PeerConnection newPeer = context.getBean(PeerConnection.class); newPeer.setChannel(newChannel); + allowBlockFetch(newPeer); newPeer.updateFetchLatency(50L); - Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Sha256Hash hash = new BlockId(Sha256Hash.ZERO_HASH, 1); Item item = new Item(hash, InventoryType.BLOCK); newPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); @@ -159,7 +161,7 @@ public void testSwitchOnOldPeerTimeout() throws Exception { method.invoke(service, fetchBlockInfo); verify(newChannel).send(any(byte[].class)); - Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + assertTrackedPeer(newPeer); } @Test @@ -172,10 +174,11 @@ public void testSwitchOnHardTimeoutWhenOldPeerUnseeded() throws Exception { when(candidateChannel.getAvgLatency()).thenReturn(50L); ReflectUtils.setFieldValue(oldPeer, "channel", oldChannel); ReflectUtils.setFieldValue(candidate, "channel", candidateChannel); + allowBlockFetch(candidate); candidate.updateFetchLatency(50L); doNothing().when(candidateChannel).send(any(byte[].class)); - Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Sha256Hash hash = new BlockId(Sha256Hash.ZERO_HASH, 1); Item item = new Item(hash, InventoryType.BLOCK); candidate.getAdvInvReceive().put(item, System.currentTimeMillis()); when(tronNetDelegate.getActivePeer()).thenReturn(Arrays.asList(oldPeer, candidate)); @@ -194,7 +197,7 @@ public void testSwitchOnHardTimeoutWhenOldPeerUnseeded() throws Exception { method.invoke(service, fetchBlockInfo); verify(candidateChannel).send(any(byte[].class)); - Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + assertTrackedPeer(candidate); } /** @@ -225,10 +228,11 @@ public void testNoSwitchWhenOldPeerLatencyUnknown() throws Exception { PeerConnection newPeer = context.getBean(PeerConnection.class); newPeer.setChannel(newChannel); + allowBlockFetch(newPeer); // first real sample replaces the unsampled state, then clamps 999 to 500 newPeer.updateFetchLatency(999L); - Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Sha256Hash hash = new BlockId(Sha256Hash.ZERO_HASH, 1); Item item = new Item(hash, InventoryType.BLOCK); oldPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); @@ -286,8 +290,9 @@ public void testSwitchWhenBothPeersUnseededReadsChannelFallback() throws Excepti oldPeer.setChannel(oldChannel); PeerConnection newPeer = context.getBean(PeerConnection.class); newPeer.setChannel(newChannel); + allowBlockFetch(newPeer); - Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Sha256Hash hash = new BlockId(Sha256Hash.ZERO_HASH, 1); Item item = new Item(hash, InventoryType.BLOCK); oldPeer.getAdvInvReceive().put(item, System.currentTimeMillis()); @@ -314,7 +319,7 @@ public void testSwitchWhenBothPeersUnseededReadsChannelFallback() throws Excepti // failover: unsampled candidate reads channel fallback 0 and wins the comparison verify(newChannel).send(any(byte[].class)); - Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + assertTrackedPeer(newPeer); } /** @@ -332,9 +337,10 @@ public void testSwitchToUnseededCandidateWhenOldPeerSeeded() throws Exception { ReflectUtils.setFieldValue(oldPeer, "channel", oldChannel); oldPeer.updateFetchLatency(200L); ReflectUtils.setFieldValue(candidate, "channel", candidateChannel); + allowBlockFetch(candidate); doNothing().when(candidateChannel).send(any(byte[].class)); - Sha256Hash hash = Sha256Hash.wrap(ByteString.copyFrom(new byte[32])); + Sha256Hash hash = new BlockId(Sha256Hash.ZERO_HASH, 1); Item item = new Item(hash, InventoryType.BLOCK); candidate.getAdvInvReceive().put(item, System.currentTimeMillis()); when(tronNetDelegate.getActivePeer()).thenReturn(Arrays.asList(oldPeer, candidate)); @@ -351,7 +357,7 @@ public void testSwitchToUnseededCandidateWhenOldPeerSeeded() throws Exception { method.invoke(service, info); verify(candidateChannel).send(any(byte[].class)); - Assert.assertNull(ReflectUtils.getFieldObject(service, "fetchBlockInfo")); + assertTrackedPeer(candidate); } /** @@ -476,4 +482,15 @@ public void testFetchLatencyIsIsolatedAcrossConnections() { // not the first connection's estimate (500) and not a cross-connection value Assert.assertEquals(50L, second.getFetchLatency()); } + + private void assertTrackedPeer(PeerConnection expected) { + Object fetchBlockInfo = ReflectUtils.getFieldObject(service, "fetchBlockInfo"); + Assert.assertNotNull(fetchBlockInfo); + Assert.assertSame(expected, ReflectUtils.getFieldObject(fetchBlockInfo, "peer")); + } + + private void allowBlockFetch(PeerConnection peer) { + peer.setNeedSyncFromPeer(false); + peer.setNeedSyncFromUs(false); + } } diff --git a/framework/src/test/java/org/tron/core/net/services/RelayServiceTest.java b/framework/src/test/java/org/tron/core/net/services/RelayServiceTest.java index 6a9c07010c9..d833a9c526a 100644 --- a/framework/src/test/java/org/tron/core/net/services/RelayServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/RelayServiceTest.java @@ -452,7 +452,10 @@ private void assertHelloAuthentication(RelayService relay, long timestamp, ByteS Channel channel = mock(Channel.class); Mockito.when(channel.getInetSocketAddress()).thenReturn(socketAddress); Mockito.when(channel.getInetAddress()).thenReturn(socketAddress.getAddress()); + Protocol.HelloMessage.BlockId blockId = Protocol.HelloMessage.BlockId.newBuilder() + .setHash(Sha256Hash.ZERO_HASH.getByteString()).build(); HelloMessage message = new HelloMessage(Protocol.HelloMessage.newBuilder() + .setGenesisBlockId(blockId).setSolidBlockId(blockId).setHeadBlockId(blockId) .setTimestamp(timestamp).setAddress(address).setSignature(signature).build().toByteArray()); Assert.assertFalse(TronNetService.getP2pConfig().getTrustNodes() diff --git a/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java b/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java index d518d7dcc7b..d2e81068f6c 100644 --- a/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java +++ b/framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java @@ -243,8 +243,9 @@ public void testProcessingBlockResumesThrottledFetch() throws Exception { syncingPeer.setNeedSyncFromPeer(true); syncingPeer.setFetchAble(true); Mockito.when(delegate.getActivePeer()).thenReturn(Collections.singletonList(syncingPeer)); - Mockito.when(delegate.getSolidBlockId()) - .thenReturn(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 0)); + BlockCapsule.BlockId head = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 0); + Mockito.when(delegate.getSolidBlockId()).thenReturn(head); + Mockito.when(delegate.getHeadBlockId()).thenReturn(head); Mockito.when(delegate.getBlockLock()).thenReturn(new Object()); BlockCapsule block = new BlockCapsule(Protocol.Block.newBuilder().setBlockHeader( Protocol.BlockHeader.newBuilder().setRawData( @@ -263,6 +264,10 @@ public void testProcessingBlockResumesThrottledFetch() throws Exception { Method handle = SyncService.class.getDeclaredMethod("handleSyncBlock"); handle.setAccessible(true); handle.invoke(sync); + Mockito.verify(delegate).processBlock(Mockito.argThat(processed -> + processed.getBlockId().equals(block.getBlockId())), Mockito.eq(true)); + Mockito.verify(channel, Mockito.never()).close(Mockito.anyLong()); + Mockito.verify(channel, Mockito.never()).close(); Assert.assertTrue("Freeing the window must wake the fetch scheduler", (boolean) ReflectUtils.getFieldObject(sync, "fetchFlag")); fetch.invoke(sync); @@ -270,6 +275,121 @@ public void testProcessingBlockResumesThrottledFetch() throws Exception { sync.close(); } + @Test + public void testUnneededBlocksReleasePendingState() throws Exception { + SyncService sync = new SyncService(); + try { + TronNetDelegate delegate = mock(TronNetDelegate.class); + ReflectUtils.setFieldValue(sync, "tronNetDelegate", delegate); + Mockito.when(delegate.getBlockLock()).thenReturn(new Object()); + Mockito.when(delegate.getSolidBlockId()) + .thenReturn(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 0)); + peer = context.getBean(PeerConnection.class); + Channel channel = mock(Channel.class); + Mockito.when(channel.getInetAddress()).thenReturn(inetSocketAddress.getAddress()); + peer.setChannel(channel); + peer.setNeedSyncFromPeer(false); + peer.setNeedSyncFromUs(false); + peer.setTronState(TronState.SYNC_COMPLETED); + Mockito.when(delegate.getActivePeer()).thenReturn(Collections.singletonList(peer)); + + BlockCapsule.BlockId known = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 1); + BlockCapsule.BlockId unknown = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 2); + BlockCapsule.BlockId queued = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 3); + BlockCapsule.BlockId preceding = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 4); + Mockito.when(delegate.containBlock(known)).thenReturn(true); + Mockito.when(delegate.containBlock(queued)).thenReturn(true); + peer.getSyncBlockToFetch().add(preceding); + peer.getSyncBlockToFetch().add(queued); + peer.getSyncBlockInProcess().add(known); + peer.getSyncBlockInProcess().add(unknown); + peer.getSyncBlockInProcess().add(queued); + Cache requests = + (Cache) + ReflectUtils.getFieldObject(sync, "requestBlockIds"); + requests.put(known, peer); + requests.put(unknown, peer); + requests.put(queued, peer); + Map waiting = (Map) + ReflectUtils.getFieldObject(sync, "blockWaitToProcess"); + waiting.put(new UnparsedBlock(known, new byte[0]), peer); + waiting.put(new UnparsedBlock(unknown, new byte[0]), peer); + waiting.put(new UnparsedBlock(queued, new byte[0]), peer); + + Method handle = SyncService.class.getDeclaredMethod("handleSyncBlock"); + handle.setAccessible(true); + handle.invoke(sync); + + Assert.assertEquals(1, waiting.size()); + Assert.assertFalse(waiting.containsKey(new UnparsedBlock(known, new byte[0]))); + Assert.assertFalse(waiting.containsKey(new UnparsedBlock(unknown, new byte[0]))); + Assert.assertTrue(waiting.containsKey(new UnparsedBlock(queued, new byte[0]))); + Assert.assertFalse(peer.getSyncBlockInProcess().contains(known)); + Assert.assertFalse(peer.getSyncBlockInProcess().contains(unknown)); + Assert.assertTrue(peer.getSyncBlockInProcess().contains(queued)); + Assert.assertNull(requests.getIfPresent(known)); + Assert.assertNull(requests.getIfPresent(unknown)); + Assert.assertSame(peer, requests.getIfPresent(queued)); + Assert.assertTrue((boolean) ReflectUtils.getFieldObject(sync, "fetchFlag")); + + peer.getSyncBlockToFetch().clear(); + handle.invoke(sync); + Assert.assertTrue(waiting.isEmpty()); + Assert.assertTrue(peer.isBlockFetchIdle()); + Assert.assertNull(requests.getIfPresent(queued)); + } finally { + sync.close(); + } + } + + @Test + public void testDeserializeFailureAllowsSyncBlockRetry() throws Exception { + SyncService sync = new SyncService(); + try { + TronNetDelegate delegate = mock(TronNetDelegate.class); + ReflectUtils.setFieldValue(sync, "tronNetDelegate", delegate); + Mockito.when(delegate.getBlockLock()).thenReturn(new Object()); + Mockito.when(delegate.getSolidBlockId()) + .thenReturn(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 0)); + PeerConnection provider = new PeerConnection(); + Channel channel = mock(Channel.class); + Mockito.when(channel.getInetAddress()).thenReturn(inetSocketAddress.getAddress()); + provider.setChannel(channel); + provider.setNeedSyncFromPeer(true); + provider.setFetchAble(true); + Mockito.when(delegate.getActivePeer()).thenReturn(Collections.singletonList(provider)); + BlockCapsule.BlockId blockId = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 1); + provider.getSyncBlockToFetch().add(blockId); + provider.getSyncBlockInProcess().add(blockId); + Cache requests = + (Cache) + ReflectUtils.getFieldObject(sync, "requestBlockIds"); + requests.put(blockId, provider); + Map waiting = (Map) + ReflectUtils.getFieldObject(sync, "blockWaitToProcess"); + waiting.put(new UnparsedBlock(blockId, new byte[]{(byte) 0x80}), provider); + + Method handle = SyncService.class.getDeclaredMethod("handleSyncBlock"); + handle.setAccessible(true); + handle.invoke(sync); + + Assert.assertTrue(waiting.isEmpty()); + Assert.assertTrue(provider.getSyncBlockInProcess().isEmpty()); + Assert.assertNull(requests.getIfPresent(blockId)); + Assert.assertEquals(Collections.singletonList(blockId), + new ArrayList<>(provider.getSyncBlockToFetch())); + Assert.assertTrue((boolean) ReflectUtils.getFieldObject(sync, "fetchFlag")); + + Method fetch = SyncService.class.getDeclaredMethod("startFetchSyncBlock"); + fetch.setAccessible(true); + fetch.invoke(sync); + Assert.assertTrue(provider.getSyncBlockRequested().containsKey(blockId)); + Assert.assertSame(provider, requests.getIfPresent(blockId)); + } finally { + sync.close(); + } + } + @Test public void sameAddressPeersKeepSyncRequestsAndResponsesIndependent() throws Exception { InetSocketAddress address = new InetSocketAddress("127.0.0.2", 10001);