Skip advanced block in NodeMessageHandler

rskj-core/src/main/java/co/rsk/net/BlockProcessor.java

@@ -51,6 +51,8 @@ public interface BlockProcessor {
 
     boolean hasBetterBlockToSync();
 
+    boolean isAdvancedBlock(long number);
+
     // New messages for RSK's sync protocol
 
     void processBlockRequest(MessageChannel sender, long requestId, byte[] hash);

rskj-core/src/main/java/co/rsk/net/NodeBlockProcessor.java

@@ -73,12 +73,28 @@ public NodeBlockProcessor(
         this.blockSyncService = blockSyncService;
         this.syncConfiguration = syncConfiguration;
     }
+
     @Override
     @Nonnull
     public Blockchain getBlockchain() {
         return this.blockchain;
     }
 
+    /**
+     * Detect a block number that is too advanced
+     * based on sync chunk size and maximum number of chuncks
+     *
+     * @param blockNumber   the block number to check
+     * @return  true if the block number is too advanced
+     */
+    @Override
+    public boolean isAdvancedBlock(long blockNumber) {
+        int syncMaxDistance = syncConfiguration.getChunkSize() * syncConfiguration.getMaxSkeletonChunks();
+        long bestBlockNumber = this.getBestBlockNumber();
+
+        return blockNumber > bestBlockNumber + syncMaxDistance;
+    }
+
     /**
      * processNewBlockHashesMessage processes a "NewBlockHashes" message. This means that we received hashes
      * from new blocks and we should request all the blocks that we don't have.

rskj-core/src/main/java/co/rsk/net/NodeMessageHandler.java

@@ -282,17 +282,25 @@ private boolean isValidBlock(@Nonnull final Block block) {
      */
     private void processBlockMessage(@Nonnull final MessageChannel sender, @Nonnull final BlockMessage message) {
         final Block block = message.getBlock();
+
         logger.trace("Process block {} {}", block.getNumber(), block.getShortHash());
 
         if (block.isGenesis()) {
             logger.trace("Skip block processing {} {}", block.getNumber(), block.getShortHash());
             return;
         }
 
+        long blockNumber = block.getNumber();
+
+        if (this.blockProcessor.isAdvancedBlock(blockNumber)) {
+            logger.trace("Too advanced block {} {}", blockNumber, block.getShortHash());
+            return;
+        }
+
         Metrics.processBlockMessage("start", block, sender.getPeerNodeID());
 
         if (!isValidBlock(block)) {
-            logger.trace("Invalid block {} {}", block.getNumber(), block.getShortHash());
+            logger.trace("Invalid block {} {}", blockNumber, block.getShortHash());
             recordEvent(sender, EventType.INVALID_BLOCK);
             return;
         }

rskj-core/src/main/java/co/rsk/validators/BlockCompositeRule.java

@@ -47,10 +47,11 @@ public BlockCompositeRule(BlockValidationRule... rules) {
     @Override
     public boolean isValid(Block block) {
         String shortHash = block.getShortHash();
-        logger.debug("Validating block {}", shortHash);
+        long number = block.getNumber();
+        logger.debug("Validating block {} {}", shortHash, number);
         for(BlockValidationRule rule : this.rules) {
             if(!rule.isValid(block)) {
-                logger.warn("Error Validating block {}", shortHash);
+                logger.warn("Error Validating block {} {}", shortHash, number);
                 return false;
             }
         }

rskj-core/src/main/java/co/rsk/validators/BlockParentCompositeRule.java

@@ -48,10 +48,11 @@ public BlockParentCompositeRule(BlockParentDependantValidationRule... rules) {
     @Override
     public boolean isValid(Block block, Block parent) {
         final String shortHash = block.getShortHash();
-        logger.debug("Validating block {}", shortHash);
+        long number = block.getNumber();
+        logger.debug("Validating block {} {}", shortHash, number);
         for(BlockParentDependantValidationRule rule : this.rules) {
             if(!rule.isValid(block, parent)) {
-                logger.warn("Error Validating block {}", shortHash);
+                logger.warn("Error Validating block {} {}", shortHash, number);
                 return false;
             }
         }

rskj-core/src/test/java/co/rsk/net/NodeBlockProcessorTest.java

@@ -83,6 +83,24 @@ public void processBlockWithTooMuchHeight() throws UnknownHostException {
         Assert.assertEquals(0, store.size());
     }
 
+    @Test
+    public void advancedBlock() throws UnknownHostException {
+        final BlockStore store = new BlockStore();
+        final MessageChannel sender = new SimpleMessageChannel();
+
+        final BlockNodeInformation nodeInformation = new BlockNodeInformation();
+        final SyncConfiguration syncConfiguration = SyncConfiguration.IMMEDIATE_FOR_TESTING;
+
+        final Blockchain blockchain = BlockChainBuilder.ofSize(0);
+        final long advancedBlockNumber = syncConfiguration.getChunkSize() * syncConfiguration.getMaxSkeletonChunks() + blockchain.getBestBlock().getNumber() + 1;
+
+        BlockSyncService blockSyncService = new BlockSyncService(store, blockchain, nodeInformation, syncConfiguration);
+        final NodeBlockProcessor processor = new NodeBlockProcessor(store, blockchain, nodeInformation, blockSyncService, syncConfiguration);
+
+        Assert.assertTrue(processor.isAdvancedBlock(advancedBlockNumber));
+        Assert.assertFalse(processor.isAdvancedBlock(advancedBlockNumber - 1));
+    }
+
     @Test
     public void processBlockAddingToBlockchain() {
         Blockchain blockchain = BlockChainBuilder.ofSize(10);

rskj-core/src/test/java/co/rsk/net/NodeMessageHandlerTest.java

@@ -129,6 +129,28 @@ public void skipProcessGenesisBlock() throws UnknownHostException {
         Assert.assertTrue(pscoring.isEmpty());
     }
 
+    @Test
+    public void skipAdvancedBlock() throws UnknownHostException {
+        SimpleMessageChannel sender = new SimpleMessageChannel();
+        PeerScoringManager scoring = createPeerScoringManager();
+        SimpleBlockProcessor sbp = new SimpleBlockProcessor();
+        sbp.setBlockGap(100000);
+        NodeMessageHandler processor = new NodeMessageHandler(config, sbp, null, null, null,null, scoring, new DummyBlockValidationRule());
+        Block block = new BlockGenerator().createBlock(200000, 0);
+        Message message = new BlockMessage(block);
+
+        processor.processMessage(sender, message);
+
+        Assert.assertNotNull(sbp.getBlocks());
+        Assert.assertEquals(0, sbp.getBlocks().size());
+        Assert.assertTrue(scoring.isEmpty());
+
+        PeerScoring pscoring = scoring.getPeerScoring(sender.getPeerNodeID());
+
+        Assert.assertNotNull(pscoring);
+        Assert.assertTrue(pscoring.isEmpty());
+    }
+
     @Test
     public void postBlockMessageTwice() throws InterruptedException, UnknownHostException {
         MessageChannel sender = new SimpleMessageChannel();

rskj-core/src/test/java/co/rsk/net/simples/SimpleBlockProcessor.java

@@ -44,6 +44,7 @@ public class SimpleBlockProcessor implements BlockProcessor {
     private long requestId;
     private byte[] hash;
     private int count;
+    private long blockGap = 1000000;
 
     @Override
     public BlockProcessResult processBlock(MessageChannel sender, Block block) {
@@ -58,17 +59,26 @@ public void processGetBlock(MessageChannel sender, byte[] hash) {
 
     }
 
+    @Override
+    public boolean isAdvancedBlock(long number) {
+        return number >= this.blockGap;
+    }
+
+    public void setBlockGap(long gap) {
+        this.blockGap = gap;
+    }
+
     @Override
     public void processBlockRequest(MessageChannel sender, long requestId, byte[] hash) {
         this.requestId = requestId;
         this.hash = hash;
+        this.count = count;
     }
 
     @Override
     public void processBlockHeadersRequest(MessageChannel sender, long requestId, byte[] hash, int count) {
         this.requestId = requestId;
         this.hash = hash;
-        this.count = count;
     }
 
     @Override