fix(core-blockchain): AcceptBlockHandler apply block to tx pool before db

.github/workflows/functional.yml

@@ -824,3 +824,51 @@ jobs:
                   POSTGRES_USER: ark
                   POSTGRES_PASSWORD: password
                   POSTGRES_DB: ark_unitnet
+
+    pool:
+        runs-on: ubuntu-latest
+
+        services:
+            postgres:
+                image: postgres:10.8
+                env:
+                    POSTGRES_USER: ark
+                    POSTGRES_PASSWORD: password
+                    POSTGRES_DB: ark_unitnet
+                ports:
+                    - 5432:5432
+                options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5
+
+        strategy:
+            matrix:
+                node-version: [12.x]
+
+        steps:
+            - uses: actions/checkout@v1
+            - name: Cache node modules
+              uses: actions/cache@v1
+              with:
+                  path: node_modules
+                  key: ${{ runner.os }}-node-${{ hashFiles('**/yarn.lock') }}
+                  restore-keys: ${{ runner.os }}-node-
+            - name: Use Node.js ${{ matrix.node-version }}
+              uses: actions/setup-node@v1
+              with:
+                  node-version: ${{ matrix.node-version }}
+            - name: Update system
+              run: sudo apt-get update -y
+            - name: Install xsel & postgresql-client
+              run: sudo apt-get install -q xsel postgresql-client
+            - name: Install and build packages
+              run: yarn setup
+            - name: Create .core/database directory
+              run: mkdir -p $HOME/.core/database
+            - name: Functional tests
+              run: yarn test __tests__/functional/pool/pool.test.ts
+
+              env:
+                  CORE_DB_DATABASE: ark_unitnet
+                  CORE_DB_USERNAME: ark
+                  POSTGRES_USER: ark
+                  POSTGRES_PASSWORD: password
+                  POSTGRES_DB: ark_unitnet

__tests__/functional/pool/pool.test.ts

@@ -0,0 +1,180 @@
+import { Identities, Managers, Utils } from "@arkecosystem/crypto";
+import delay from "delay";
+import { TransactionFactory } from "../../helpers/transaction-factory";
+import { secrets } from "../../utils/config/testnet/delegates.json";
+import * as support from "./__support__";
+
+beforeAll(async () => {
+    await support.setUp();
+    Managers.configManager.setFromPreset("testnet");
+});
+afterAll(support.tearDown);
+
+describe("applyToRecipient - Multipayment scenario", () => {
+    /*
+     * Scenario :
+     * - init bob and alice wallet
+     * - send an initial tx from bob to index his wallet in tx pool
+     * - send a multipayment from alice including bob in payment recipients
+     * - send bob funds received from multipayment to a random address
+     * - this last transaction from bob fails if pool wallet is not updated correctly by multipayment tx
+     */
+    const bobPassphrase = "bob pass phrase1";
+    const bobAddress = Identities.Address.fromPassphrase(bobPassphrase, 23);
+    const bobInitialFund = 50 * 1e8; // 50 ARK
+
+    const alicePassphrase = "alice pass phrase1";
+    const aliceAddress = Identities.Address.fromPassphrase(alicePassphrase, 23);
+    const aliceInitialFund = 2500 * 1e8; // 2500 ARK
+
+    const randomAddress = Identities.Address.fromPassphrase("ran dom addr1", 23);
+
+    it("should correctly update recipients pool wallet balance after a multipayment", async () => {
+        const initialTxToBob = TransactionFactory.transfer(bobAddress, bobInitialFund)
+            .withPassphrase(secrets[1])
+            .createOne();
+        const initialTxToAlice = TransactionFactory.transfer(aliceAddress, aliceInitialFund)
+            .withPassphrase(secrets[2])
+            .createOne();
+        await expect(initialTxToBob).toBeAccepted();
+        await support.forge([initialTxToBob, initialTxToAlice]);
+        await delay(1000);
+
+        const initialTxFromBob = TransactionFactory.transfer(bobAddress, 1)
+            .withPassphrase(bobPassphrase)
+            .createOne();
+        await expect(initialTxFromBob).toBeAccepted();
+        await support.forge([initialTxFromBob]);
+        await delay(1000);
+
+        const multipaymentToBobAndAlice = TransactionFactory.multiPayment([
+            {
+                recipientId: bobAddress,
+                amount: (2000 * 1e8).toFixed(), // 2000 ARK
+            },
+            {
+                recipientId: aliceAddress,
+                amount: (10 * 1e8).toFixed(), // 10 ARK
+            },
+        ])
+            .withPassphrase(alicePassphrase)
+            .createOne();
+        await support.forge([multipaymentToBobAndAlice]);
+        await delay(1000);
+        await expect(multipaymentToBobAndAlice.id).toBeForged();
+
+        const bobTransfer = TransactionFactory.transfer(randomAddress, 2000 * 1e8)
+            .withPassphrase(bobPassphrase)
+            .createOne();
+        await expect(bobTransfer).toBeAccepted();
+        await support.forge([bobTransfer]);
+        await delay(1000);
+    });
+});
+
+describe("applyToRecipient - transfer and multipayment classic scenarios", () => {
+    it("should not accept a transfer in the pool with more amount than sender balance", async () => {
+        // just send funds to a new wallet, and try to send more than the funds from this new wallet
+        const bobPassphrase = "bob pass phrase2";
+        const bobAddress = Identities.Address.fromPassphrase(bobPassphrase, 23);
+        const bobInitialFund = 100 * 1e8; // 100 ARK
+
+        const randomAddress = Identities.Address.fromPassphrase(secrets[1], 23);
+        const initialTxToBob = TransactionFactory.transfer(bobAddress, bobInitialFund)
+            .withPassphrase(secrets[1])
+            .createOne();
+
+        await support.forge([initialTxToBob]);
+        await delay(1000);
+
+        // the fees for this are making the transfer worth more than bob balance
+        const bobTransferMoreThanBalance = TransactionFactory.transfer(randomAddress, bobInitialFund)
+            .withPassphrase(bobPassphrase)
+            .createOne();
+        await expect(bobTransferMoreThanBalance).not.toBeAccepted();
+
+        // now a transaction with fees + amount === balance should pass
+        const fee = 1e7;
+        const bobTransferValid = TransactionFactory.transfer(randomAddress, bobInitialFund - fee)
+            .withPassphrase(bobPassphrase)
+            .withFee(fee)
+            .createOne();
+        await expect(bobTransferValid).toBeAccepted();
+        await delay(1000);
+    });
+
+    it("should not accept a transfer in the pool with more amount than sender balance", async () => {
+        // just send funds to a new wallet with multipayment, and try to send more than the funds from this new wallet
+        const bobPassphrase = "bob pass phrase3";
+        const bobAddress = Identities.Address.fromPassphrase(bobPassphrase, 23);
+        const bobInitialFund = 100 * 1e8; // 100 ARK
+
+        const randomAddress = Identities.Address.fromPassphrase("a b c", 23);
+
+        const initialTxToBob = TransactionFactory.multiPayment([
+            {
+                recipientId: bobAddress,
+                amount: bobInitialFund.toFixed(),
+            },
+            {
+                recipientId: randomAddress,
+                amount: bobInitialFund.toFixed(),
+            },
+        ])
+            .withPassphrase(secrets[1])
+            .createOne();
+
+        await support.forge([initialTxToBob]);
+        await delay(1000);
+
+        // the fees for this are making the transfer worth more than bob balance
+        const bobTransferMoreThanBalance = TransactionFactory.transfer(randomAddress, bobInitialFund)
+            .withPassphrase(bobPassphrase)
+            .createOne();
+        await expect(bobTransferMoreThanBalance).not.toBeAccepted();
+
+        // now a transaction with fees + amount === balance should pass
+        const fee = 1e7;
+        const bobTransferValid = TransactionFactory.transfer(randomAddress, bobInitialFund - fee)
+            .withPassphrase(bobPassphrase)
+            .withFee(fee)
+            .createOne();
+        await expect(bobTransferValid).toBeAccepted();
+        await delay(1000);
+    });
+});
+
+describe("Pool transactions when AcceptBlockHandler fails", () => {
+    // just send funds to a new wallet, and try to send more than the funds from this new wallet
+    const bobPassphrase = "bob pass phrase4";
+    const bobAddress = Identities.Address.fromPassphrase(bobPassphrase, 23);
+    const bobInitialFund = 100 * 1e8; // 100 ARK
+
+    const randomAddress = Identities.Address.fromPassphrase(secrets[1], 23);
+
+    it("should keep transactions in the pool after AcceptBlockHandler fails to accept a block", async () => {
+        const initialTxToBob = TransactionFactory.transfer(bobAddress, bobInitialFund)
+            .withPassphrase(secrets[1])
+            .createOne();
+
+        await support.forge([initialTxToBob]);
+        await delay(1000);
+
+        // a valid tx to accept in the pool
+        const bobTransfer = TransactionFactory.transfer(randomAddress, 100)
+            .withPassphrase(bobPassphrase)
+            .createOne();
+        await expect(bobTransfer).toBeAccepted();
+        await expect(bobTransfer).toBeUnconfirmed();
+
+        // this one will make AcceptBlockHandler fail to accept the block
+        const bobBusinessResignation = TransactionFactory.businessResignation()
+            .withPassphrase(bobPassphrase)
+            .withNonce(Utils.BigNumber.ZERO)
+            .createOne();
+        await support.forge([bobBusinessResignation]);
+        await delay(1000);
+
+        await expect(bobTransfer).toBeUnconfirmed();
+    });
+});

__tests__/unit/core-blockchain/mocks/transactionPool.ts

@@ -2,4 +2,10 @@ export const transactionPool = {
     buildWallets: () => undefined,
     acceptChainedBlock: () => undefined,
     removeTransactionsById: () => undefined,
+    flush: () => undefined,
+    getAllTransactions: () => [],
+    addTransactions: () => undefined,
+    walletManager: {
+        reset: () => undefined,
+    },
 };

__tests__/unit/core-transaction-pool/__stubs__/connection.ts

@@ -69,6 +69,10 @@ export class Connection implements TransactionPool.IConnection {
         return;
     }
 
+    public getAllTransactions(): Interfaces.ITransaction[] {
+        return [];
+    }
+
     public async getTransactionsForForging(blockSize: number): Promise<string[]> {
         return [];
     }

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

@@ -1,28 +1,34 @@
+import { TransactionPool } from "@arkecosystem/core-interfaces";
 import { BlockProcessorResult } from "../block-processor";
 import { BlockHandler } from "./block-handler";
 
 export class AcceptBlockHandler extends BlockHandler {
     public async execute(): Promise<BlockProcessorResult> {
         const { database, state, transactionPool } = this.blockchain;
 
+        let transactionPoolWasReset: boolean = false;
         try {
-            await database.applyBlock(this.block);
-
-            // Check if we recovered from a fork
-            if (state.forkedBlock && state.forkedBlock.data.height === this.block.data.height) {
-                this.logger.info("Successfully recovered from fork");
-                state.forkedBlock = undefined;
-            }
-
             if (transactionPool) {
                 try {
                     await transactionPool.acceptChainedBlock(this.block);
                 } catch (error) {
+                    // reset transaction pool as it could be out of sync with db state
+                    await this.resetTransactionPool(transactionPool);
+                    transactionPoolWasReset = true;
+
                     this.logger.warn("Issue applying block to transaction pool");
                     this.logger.debug(error.stack);
                 }
             }
 
+            await database.applyBlock(this.block);
+
+            // Check if we recovered from a fork
+            if (state.forkedBlock && state.forkedBlock.data.height === this.block.data.height) {
+                this.logger.info("Successfully recovered from fork");
+                state.forkedBlock = undefined;
+            }
+
             // Reset wake-up timer after chaining a block, since there's no need to
             // wake up at all if blocks arrive periodically. Only wake up when there are
             // no new blocks.
@@ -39,10 +45,25 @@ export class AcceptBlockHandler extends BlockHandler {
 
             return BlockProcessorResult.Accepted;
         } catch (error) {
+            if (transactionPool && !transactionPoolWasReset) {
+                // reset transaction pool as it could be out of sync with db state
+                await this.resetTransactionPool(transactionPool);
+            }
+
             this.logger.warn(`Refused new block ${JSON.stringify(this.block.data)}`);
             this.logger.debug(error.stack);
 
             return super.execute();
         }
     }
+
+    private async resetTransactionPool(transactionPool: TransactionPool.IConnection): Promise<void> {
+        // backup transactions from pool, flush it, reset wallet manager, re-add transactions
+        const transactions = transactionPool.getAllTransactions();
+
+        transactionPool.flush();
+        transactionPool.walletManager.reset();
+
+        await transactionPool.addTransactions(transactions);
+    }
 }