fix(core-blockchain): take milestones into account when downloading blocks in parallel

__tests__/unit/core-blockchain/blockchain.test.ts

@@ -85,8 +85,9 @@ describe("Blockchain", () => {
         });
 
         it("should enqueue the blocks provided", async () => {
-            const processQueuePush = jest.spyOn(blockchain.queue, "push");
+            blockchain.state.lastDownloadedBlock = blocks101to155[54];
 
+            const processQueuePush = jest.spyOn(blockchain.queue, "push");
             const blocksToEnqueue = [blocks101to155[54]];
             blockchain.enqueueBlocks(blocksToEnqueue);
             expect(processQueuePush).toHaveBeenCalledWith({ blocks: blocksToEnqueue });
@@ -179,9 +180,10 @@ describe("Blockchain", () => {
                 state: { maxLastBlocks: 50 },
             });
 
-            blockchain.state.setLastBlock(genesisBlock);
+            const block = Blocks.BlockFactory.fromData(blocks2to100[0]);
+            blockchain.state.setLastBlock(block);
 
-            expect(blockchain.getLastBlock()).toEqual(genesisBlock);
+            expect(blockchain.getLastBlock()).toEqual(block);
         });
     });
 
@@ -269,11 +271,9 @@ describe("Blockchain", () => {
             it("should be ok", () => {
                 expect(
                     blockchain.isSynced({
-                        data: {
-                            timestamp: Crypto.Slots.getTime(),
-                            height: genesisBlock.height,
-                        },
-                    } as Interfaces.IBlock),
+                        timestamp: Crypto.Slots.getTime(),
+                        height: genesisBlock.height,
+                    } as Interfaces.IBlockData),
                 ).toBeTrue();
             });
         });

__tests__/unit/core-blockchain/state-machine.test.ts

@@ -299,7 +299,7 @@ describe("State Machine", () => {
             });
 
             beforeEach(() => {
-                stateStorage.lastDownloadedBlock = BlockFactory.fromData(genesisBlock);
+                stateStorage.lastDownloadedBlock = genesisBlock;
             });
 
             afterEach(() => jest.resetAllMocks());

__tests__/unit/core-blockchain/stubs/state-storage.ts

@@ -4,7 +4,7 @@ import { Blocks, Interfaces } from "@arkecosystem/crypto";
 
 export class StateStoreStub implements State.IStateStore {
     public blockchain: any;
-    public lastDownloadedBlock: Interfaces.IBlock | undefined;
+    public lastDownloadedBlock: Interfaces.IBlockData | undefined;
     public blockPing: any;
     public started: boolean;
     public forkedBlock: Interfaces.IBlock | undefined;
@@ -39,7 +39,7 @@ export class StateStoreStub implements State.IStateStore {
     }
 
     public getLastBlock(): Interfaces.IBlock | undefined {
-        return this.lastDownloadedBlock || undefined;
+        return Blocks.BlockFactory.fromData(this.lastDownloadedBlock) || undefined;
     }
 
     public getLastBlockIds(): string[] {
@@ -72,7 +72,7 @@ export class StateStoreStub implements State.IStateStore {
     public reset(): void {}
 
     public setLastBlock(block: Blocks.Block): void {
-        this.lastDownloadedBlock = block;
+        this.lastDownloadedBlock = block.data;
     }
 }
 

__tests__/unit/core-database/__fixtures__/state-storage-stub.ts

@@ -4,7 +4,7 @@ import { Blocks, Interfaces } from "@arkecosystem/crypto";
 
 export class StateStoreStub implements State.IStateStore {
     public blockchain: any;
-    public lastDownloadedBlock: Interfaces.IBlock | undefined;
+    public lastDownloadedBlock: Interfaces.IBlockData | undefined;
     public blockPing: any;
     public started: boolean;
     public forkedBlock: Interfaces.IBlock | undefined;

__tests__/unit/core-state/__fixtures__/state-storage-stub.ts

@@ -4,7 +4,7 @@ import { Blocks, Interfaces } from "@arkecosystem/crypto";
 
 export class StateStorageStub implements State.IStateStorage {
     public blockchain: any;
-    public lastDownloadedBlock: Interfaces.IBlock | undefined;
+    public lastDownloadedBlock: Interfaces.IBlockData | undefined;
     public blockPing: any;
     public started: boolean;
     public forkedBlock: Interfaces.IBlock | undefined;

packages/core-blockchain/src/blockchain.ts

@@ -10,7 +10,7 @@ import {
     State,
     TransactionPool,
 } from "@arkecosystem/core-interfaces";
-import { Blocks, Crypto, Interfaces } from "@arkecosystem/crypto";
+import { Blocks, Crypto, Interfaces, Managers } from "@arkecosystem/crypto";
 
 import async from "async";
 import delay from "delay";
@@ -206,7 +206,7 @@ export class Blockchain implements blockchain.IBlockchain {
      * @return {void}
      */
     public clearAndStopQueue(): void {
-        this.state.lastDownloadedBlock = this.getLastBlock();
+        this.state.lastDownloadedBlock = this.getLastBlock().data;
 
         this.queue.pause();
         this.clearQueue();
@@ -263,6 +263,13 @@ export class Blockchain implements blockchain.IBlockchain {
             return;
         }
 
+        const lastDownloadedHeight: number = this.getLastDownloadedBlock().height;
+        const milestoneHeights: number[] = Managers.configManager
+            .getMilestones()
+            .map(milestone => milestone.height)
+            .sort((a, b) => a - b)
+            .filter(height => height >= lastDownloadedHeight);
+
         // divide blocks received into chunks depending on number of transactions
         // this is to avoid blocking the application when processing "heavy" blocks
         let currentBlocksChunk = [];
@@ -271,15 +278,17 @@ export class Blockchain implements blockchain.IBlockchain {
             currentBlocksChunk.push(block);
             currentTransactionsCount += block.numberOfTransactions;
 
-            if (currentTransactionsCount >= 150 || currentBlocksChunk.length > 100) {
+            const nextMilestone = milestoneHeights[0] && milestoneHeights[0] === block.height;
+            if (currentTransactionsCount >= 150 || currentBlocksChunk.length > 100 || nextMilestone) {
                 this.queue.push({ blocks: currentBlocksChunk });
                 currentBlocksChunk = [];
                 currentTransactionsCount = 0;
+                milestoneHeights.shift();
             }
         }
         this.queue.push({ blocks: currentBlocksChunk });
 
-        this.state.lastDownloadedBlock = BlockFactory.fromData(blocks.slice(-1)[0]);
+        this.state.lastDownloadedBlock = blocks.slice(-1)[0];
     }
 
     /**
@@ -311,7 +320,7 @@ export class Blockchain implements blockchain.IBlockchain {
             const newLastBlock = BlockFactory.fromData(blocksToRemove.pop());
 
             this.state.setLastBlock(newLastBlock);
-            this.state.lastDownloadedBlock = newLastBlock;
+            this.state.lastDownloadedBlock = newLastBlock.data;
         };
 
         // tslint:disable-next-line:variable-name
@@ -335,7 +344,7 @@ export class Blockchain implements blockchain.IBlockchain {
         const resetHeight: number = lastBlock.data.height - nblocks;
         logger.info(`Removing ${pluralize("block", nblocks, true)}. Reset to height ${resetHeight.toLocaleString()}`);
 
-        this.state.lastDownloadedBlock = lastBlock;
+        this.state.lastDownloadedBlock = lastBlock.data;
 
         await __removeBlocks(nblocks);
 
@@ -429,7 +438,7 @@ export class Blockchain implements blockchain.IBlockchain {
      * Reset the last downloaded block to last chained block.
      */
     public resetLastDownloadedBlock(): void {
-        this.state.lastDownloadedBlock = this.getLastBlock();
+        this.state.lastDownloadedBlock = this.getLastBlock().data;
     }
 
     /**
@@ -458,14 +467,14 @@ export class Blockchain implements blockchain.IBlockchain {
     /**
      * Determine if the blockchain is synced.
      */
-    public isSynced(block?: Interfaces.IBlock): boolean {
+    public isSynced(block?: Interfaces.IBlockData): boolean {
         if (!this.p2p.getStorage().hasPeers()) {
             return true;
         }
 
-        block = block || this.getLastBlock();
+        block = block || this.getLastBlock().data;
 
-        return Crypto.Slots.getTime() - block.data.timestamp < 3 * config.getMilestone(block.data.height).blocktime;
+        return Crypto.Slots.getTime() - block.timestamp < 3 * config.getMilestone(block.height).blocktime;
     }
 
     public async replay(targetHeight?: number): Promise<void> {
@@ -489,7 +498,7 @@ export class Blockchain implements blockchain.IBlockchain {
     /**
      * Get the last downloaded block of the blockchain.
      */
-    public getLastDownloadedBlock(): Interfaces.IBlock {
+    public getLastDownloadedBlock(): Interfaces.IBlockData {
         return this.state.lastDownloadedBlock;
     }
 

packages/core-blockchain/src/processor/handlers/accept-block-handler.ts

@@ -33,8 +33,8 @@ export class AcceptBlockHandler extends BlockHandler {
             state.setLastBlock(this.block);
 
             // Ensure the lastDownloadedBlock is never behind the last accepted block.
-            if (state.lastDownloadedBlock && state.lastDownloadedBlock.data.height < this.block.data.height) {
-                state.lastDownloadedBlock = this.block;
+            if (state.lastDownloadedBlock && state.lastDownloadedBlock.height < this.block.data.height) {
+                state.lastDownloadedBlock = this.block.data;
             }
 
             return BlockProcessorResult.Accepted;

packages/core-blockchain/src/state-machine.ts

@@ -152,7 +152,7 @@ blockchainMachine.actionMap = (blockchain: Blockchain) => ({
              *  state machine data init      *
              ******************************* */
             stateStorage.setLastBlock(block);
-            stateStorage.lastDownloadedBlock = block;
+            stateStorage.lastDownloadedBlock = block.data;
 
             // Delete all rounds from the future due to shutdown before processBlocks finished writing the blocks.
             const roundInfo = roundCalculator.calculateRound(block.data.height);
@@ -195,23 +195,24 @@ blockchainMachine.actionMap = (blockchain: Blockchain) => ({
     },
 
     async downloadBlocks() {
-        const lastDownloadedBlock: Interfaces.IBlock = stateStorage.lastDownloadedBlock || stateStorage.getLastBlock();
+        const lastDownloadedBlock: Interfaces.IBlockData =
+            stateStorage.lastDownloadedBlock || stateStorage.getLastBlock().data;
         const blocks: Interfaces.IBlockData[] = await blockchain.p2p
             .getMonitor()
-            .syncWithNetwork(lastDownloadedBlock.data.height);
+            .syncWithNetwork(lastDownloadedBlock.height);
 
         if (blockchain.isStopped) {
             return;
         }
 
         // Could have changed since entering this function, e.g. due to a rollback.
-        if (lastDownloadedBlock.data.id !== stateStorage.lastDownloadedBlock.data.id) {
+        if (lastDownloadedBlock.id !== stateStorage.lastDownloadedBlock.id) {
             return;
         }
 
         const empty: boolean = !blocks || blocks.length === 0;
         const chained: boolean =
-            !empty && (isBlockChained(lastDownloadedBlock.data, blocks[0]) || Utils.isException(blocks[0]));
+            !empty && (isBlockChained(lastDownloadedBlock, blocks[0]) || Utils.isException(blocks[0]));
 
         if (chained) {
             logger.info(
@@ -243,14 +244,14 @@ blockchainMachine.actionMap = (blockchain: Blockchain) => ({
                 logger.info("No new block found on this peer");
             } else {
                 logger.warn(`Downloaded block not accepted: ${JSON.stringify(blocks[0])}`);
-                logger.warn(`Last downloaded block: ${JSON.stringify(lastDownloadedBlock.data)}`);
+                logger.warn(`Last downloaded block: ${JSON.stringify(lastDownloadedBlock)}`);
 
                 blockchain.clearQueue();
             }
 
             if (blockchain.queue.length() === 0) {
                 stateStorage.noBlockCounter++;
-                stateStorage.lastDownloadedBlock = stateStorage.getLastBlock();
+                stateStorage.lastDownloadedBlock = stateStorage.getLastBlock().data;
             }
 
             blockchain.dispatch("NOBLOCK");

packages/core-interfaces/src/core-blockchain/blockchain.ts

@@ -107,7 +107,7 @@ export interface IBlockchain {
      * @param  {Block} [block=getLastBlock()]  block
      * @return {Boolean}
      */
-    isSynced(block?: Interfaces.IBlock): boolean;
+    isSynced(block?: Interfaces.IBlockData): boolean;
 
     /**
      * Get the last block of the blockchain.
@@ -125,7 +125,7 @@ export interface IBlockchain {
      * Get the last downloaded block of the blockchain.
      * @return {Object}
      */
-    getLastDownloadedBlock(): Interfaces.IBlock;
+    getLastDownloadedBlock(): Interfaces.IBlockData;
 
     /**
      * Get the block ping.

packages/core-interfaces/src/core-state/state-store.ts

@@ -2,7 +2,7 @@ import { Interfaces } from "@arkecosystem/crypto";
 
 export interface IStateStore {
     blockchain: any;
-    lastDownloadedBlock: Interfaces.IBlock | undefined;
+    lastDownloadedBlock: Interfaces.IBlockData | undefined;
     blockPing: any;
     started: boolean;
     forkedBlock: Interfaces.IBlock | undefined;

packages/core-p2p/src/socket-server/versions/peer.ts

@@ -66,10 +66,10 @@ export const postBlock = async ({ req }): Promise<void> => {
             return;
         }
 
-        const lastDownloadedBlock: Interfaces.IBlock = blockchain.getLastDownloadedBlock();
+        const lastDownloadedBlock: Interfaces.IBlockData = blockchain.getLastDownloadedBlock();
 
-        if (!isBlockChained(lastDownloadedBlock.data, block)) {
-            throw new UnchainedBlockError(lastDownloadedBlock.data.height, block.height);
+        if (!isBlockChained(lastDownloadedBlock, block)) {
+            throw new UnchainedBlockError(lastDownloadedBlock.height, block.height);
         }
     }
 

packages/core-state/src/stores/state.ts

@@ -13,7 +13,7 @@ import { OrderedMap, OrderedSet, Seq } from "immutable";
 export class StateStore implements State.IStateStore {
     // @TODO: make all properties private and expose them one-by-one through a getter if used outside of this class
     public blockchain: any = {};
-    public lastDownloadedBlock: Interfaces.IBlock | undefined = undefined;
+    public lastDownloadedBlock: Interfaces.IBlockData | undefined = undefined;
     public blockPing: any = undefined;
     public started: boolean = false;
     public forkedBlock: Interfaces.IBlock | undefined = undefined;

packages/crypto/src/blocks/block.ts

@@ -15,7 +15,7 @@ export class Block implements IBlock {
             error &&
             !(isException(value) || data.transactions.some((transaction: ITransactionData) => isException(transaction)))
         ) {
-            throw new BlockSchemaError(error);
+            throw new BlockSchemaError(data.height, error);
         }
 
         return value;

packages/crypto/src/errors.ts

@@ -127,8 +127,8 @@ export class MissingTransactionSignatureError extends CryptoError {
 }
 
 export class BlockSchemaError extends CryptoError {
-    constructor(what: string) {
-        super(what);
+    constructor(height: number, what: string) {
+        super(`Height (${height}): ${what}`);
     }
 }