refactor(core-transaction-pool): segregate pools by wallets

__tests__/unit/core-transaction-pool/collator.test.ts

@@ -0,0 +1,83 @@
+import { Container } from "@arkecosystem/core-kernel";
+import { Managers } from "@arkecosystem/crypto";
+
+import { Collator } from "../../../packages/core-transaction-pool/src/collator";
+
+jest.mock("@arkecosystem/crypto");
+
+describe("Collator", () => {
+    const container = new Container.Container();
+
+    describe("getBlockCandidateTransactions", () => {
+        const validator = { validate: jest.fn() };
+        const configuration = { get: jest.fn() };
+        const createTransactionValidator = jest.fn(() => validator);
+        const blockchain = { getLastBlock: jest.fn() };
+        const pool = { cleanUp: jest.fn(), removeTransaction: jest.fn() };
+        const poolQuery = { getAllFromHighestPriority: jest.fn() };
+        const logger = { error: jest.fn() };
+
+        beforeAll(() => {
+            container.unbindAll();
+            container.bind(Container.Identifiers.PluginConfiguration).toConstantValue(configuration);
+            container
+                .bind(Container.Identifiers.TransactionValidatorFactory)
+                .toConstantValue(createTransactionValidator);
+            container.bind(Container.Identifiers.BlockchainService).toConstantValue(blockchain);
+            container.bind(Container.Identifiers.TransactionPoolService).toConstantValue(pool);
+            container.bind(Container.Identifiers.TransactionPoolQuery).toConstantValue(poolQuery);
+            container.bind(Container.Identifiers.LogService).toConstantValue(logger);
+        });
+
+        beforeEach(() => {
+            validator.validate.mockClear();
+            configuration.get.mockClear();
+            createTransactionValidator.mockClear();
+            blockchain.getLastBlock.mockClear();
+            pool.cleanUp.mockClear();
+            poolQuery.getAllFromHighestPriority.mockClear();
+            logger.error.mockClear();
+        });
+
+        it("should respect milestone transaction count limit", async () => {
+            const poolTransactions = new Array(10).fill({ data: "12345678" });
+            const milestone = { block: { maxTransactions: 5 } };
+            const lastBlock = { data: { height: 10 } };
+
+            (Managers.configManager.getMilestone as jest.Mock).mockReturnValueOnce(milestone);
+            blockchain.getLastBlock.mockReturnValueOnce(lastBlock);
+            poolQuery.getAllFromHighestPriority.mockReturnValueOnce(poolTransactions);
+
+            const collator = container.resolve(Collator);
+            const candidateTransaction = await collator.getBlockCandidateTransactions();
+
+            expect(candidateTransaction.length).toBe(5);
+            expect(configuration.get).toBeCalled();
+            expect(Managers.configManager.getMilestone).toBeCalled();
+            expect(createTransactionValidator).toBeCalled();
+            expect(pool.cleanUp).toBeCalled();
+            expect(validator.validate).toBeCalledTimes(5);
+        });
+
+        it("should respect maxTransactionBytes configuration limit", async () => {
+            const poolTransactions = new Array(10).fill({ data: "12345678" });
+            const milestone = { block: { maxTransactions: 100 } };
+            const lastBlock = { data: { height: 10 } };
+
+            (Managers.configManager.getMilestone as jest.Mock).mockReturnValueOnce(milestone);
+            configuration.get.mockReturnValueOnce(25);
+            blockchain.getLastBlock.mockReturnValueOnce(lastBlock);
+            poolQuery.getAllFromHighestPriority.mockReturnValueOnce(poolTransactions);
+
+            const collator = container.resolve(Collator);
+            const candidateTransaction = await collator.getBlockCandidateTransactions();
+
+            expect(candidateTransaction.length).toBe(2);
+            expect(configuration.get).toBeCalled();
+            expect(Managers.configManager.getMilestone).toBeCalled();
+            expect(createTransactionValidator).toBeCalled();
+            expect(pool.cleanUp).toBeCalled();
+            expect(validator.validate).toBeCalledTimes(2);
+        });
+    });
+});

__tests__/unit/core-transaction-pool/processor.test.ts

@@ -1,4 +1,4 @@
-import { Container } from "@arkecosystem/core-kernel";
+import { Container, Contracts } from "@arkecosystem/core-kernel";
 import { Interfaces, Transactions } from "@arkecosystem/crypto";
 
 import { Processor } from "../../../packages/core-transaction-pool/src/processor";
@@ -9,7 +9,7 @@ describe("Processor", () => {
     const container = new Container.Container();
 
     describe("process", () => {
-        const logger = { warning: jest.fn() };
+        const logger = { warning: jest.fn(), error: jest.fn() };
         const pool = { addTransaction: jest.fn() };
         const dynamicFeeMatcher = { canEnterPool: jest.fn(), canBroadcast: jest.fn() };
         const transactionBroadcaster = { broadcastTransactions: jest.fn() };
@@ -23,30 +23,115 @@ describe("Processor", () => {
         });
 
         beforeEach(() => {
-            (Transactions.TransactionFactory.fromData as jest.Mock).mockReset();
+            (Transactions.TransactionFactory.fromData as jest.Mock).mockClear();
 
-            logger.warning.mockReset();
-            pool.addTransaction.mockReset();
-            dynamicFeeMatcher.canEnterPool.mockReset();
-            dynamicFeeMatcher.canBroadcast.mockReset();
-            transactionBroadcaster.broadcastTransactions.mockReset();
+            logger.warning.mockClear();
+            pool.addTransaction.mockClear();
+            dynamicFeeMatcher.canEnterPool.mockClear();
+            dynamicFeeMatcher.canBroadcast.mockClear();
+            transactionBroadcaster.broadcastTransactions.mockClear();
         });
 
         it("should add eligible transactions to pool", async () => {
-            (Transactions.TransactionFactory.fromData as jest.Mock).mockImplementation(d => ({ id: d.id }));
+            const data: any[] = [{ id: "id1" }, { id: "id2" }];
 
+            (Transactions.TransactionFactory.fromData as jest.Mock).mockImplementation(d => ({ id: d.id }));
             dynamicFeeMatcher.canEnterPool.mockReturnValueOnce(true).mockReturnValueOnce(false);
-            const data: any[] = [{ id: "id_eligible" }, { id: "id_non_eligible" }];
 
             const processor = container.resolve(Processor);
             await processor.process(data as Interfaces.ITransactionData[]);
 
-            expect(processor.accept).toEqual(["id_eligible"]);
-            expect(processor.invalid).toEqual(["id_non_eligible"]);
+            expect(dynamicFeeMatcher.canEnterPool).toBeCalledTimes(2);
+            expect(pool.addTransaction).toBeCalledTimes(1);
+            expect(dynamicFeeMatcher.canBroadcast).toBeCalledTimes(1);
+            expect(transactionBroadcaster.broadcastTransactions).not.toBeCalled();
+
+            expect(processor.accept).toEqual(["id1"]);
+            expect(processor.broadcast).toEqual([]);
+            expect(processor.invalid).toEqual(["id2"]);
+            expect(processor.excess).toEqual([]);
             expect(processor.errors).toEqual({
-                id_non_eligible: {
+                id2: {
                     type: "ERR_LOW_FEE",
-                    message: "Transaction id_non_eligible fee is to low to include in pool",
+                    message: "Transaction id2 fee is to low to include in pool",
+                },
+            });
+        });
+
+        it("should add broadcast eligible transaction", async () => {
+            const data: any[] = [{ id: "id1" }, { id: "id2" }];
+
+            (Transactions.TransactionFactory.fromData as jest.Mock).mockImplementation(d => ({ id: d.id }));
+            dynamicFeeMatcher.canEnterPool.mockReturnValueOnce(true).mockReturnValueOnce(true);
+            dynamicFeeMatcher.canBroadcast.mockReturnValueOnce(true).mockReturnValueOnce(false);
+
+            const processor = container.resolve(Processor);
+            await processor.process(data as Interfaces.ITransactionData[]);
+
+            expect(dynamicFeeMatcher.canEnterPool).toBeCalledTimes(2);
+            expect(pool.addTransaction).toBeCalledTimes(2);
+            expect(dynamicFeeMatcher.canBroadcast).toBeCalledTimes(2);
+            expect(transactionBroadcaster.broadcastTransactions).toBeCalled();
+
+            expect(processor.accept).toEqual(["id1", "id2"]);
+            expect(processor.broadcast).toEqual(["id1"]);
+            expect(processor.invalid).toEqual([]);
+            expect(processor.excess).toEqual([]);
+            expect(processor.errors).toEqual(undefined);
+        });
+
+        it("should rethrow unexpected error", async () => {
+            const data: any[] = [{ id: "id1" }];
+
+            (Transactions.TransactionFactory.fromData as jest.Mock).mockImplementation(d => ({ id: d.id }));
+            dynamicFeeMatcher.canEnterPool.mockReturnValueOnce(true);
+            pool.addTransaction.mockRejectedValueOnce(new Error("Unexpected error"));
+
+            const processor = container.resolve(Processor);
+            const promise = processor.process(data as Interfaces.ITransactionData[]);
+
+            await expect(promise).rejects.toThrow();
+
+            expect(dynamicFeeMatcher.canEnterPool).toBeCalledTimes(1);
+            expect(pool.addTransaction).toBeCalledTimes(1);
+            expect(dynamicFeeMatcher.canBroadcast).toBeCalledTimes(0);
+            expect(transactionBroadcaster.broadcastTransactions).not.toBeCalled();
+
+            expect(processor.accept).toEqual([]);
+            expect(processor.broadcast).toEqual([]);
+            expect(processor.invalid).toEqual(["id1"]);
+            expect(processor.excess).toEqual([]);
+            expect(processor.errors).toEqual(undefined);
+        });
+
+        it("should track excess transactions", async () => {
+            const data: any[] = [{ id: "id1" }];
+            const exceedsError = new Contracts.TransactionPool.PoolError(
+                "Exceeds",
+                "ERR_EXCEEDS_MAX_COUNT",
+                data[0] as Interfaces.ITransaction,
+            );
+
+            (Transactions.TransactionFactory.fromData as jest.Mock).mockImplementation(d => ({ id: d.id }));
+            dynamicFeeMatcher.canEnterPool.mockReturnValueOnce(true);
+            pool.addTransaction.mockRejectedValueOnce(exceedsError);
+
+            const processor = container.resolve(Processor);
+            await processor.process(data as Interfaces.ITransactionData[]);
+
+            expect(dynamicFeeMatcher.canEnterPool).toBeCalledTimes(1);
+            expect(pool.addTransaction).toBeCalledTimes(1);
+            expect(dynamicFeeMatcher.canBroadcast).toBeCalledTimes(0);
+            expect(transactionBroadcaster.broadcastTransactions).not.toBeCalled();
+
+            expect(processor.accept).toEqual([]);
+            expect(processor.broadcast).toEqual([]);
+            expect(processor.invalid).toEqual(["id1"]);
+            expect(processor.excess).toEqual(["id1"]);
+            expect(processor.errors).toEqual({
+                id1: {
+                    type: "ERR_EXCEEDS_MAX_COUNT",
+                    message: "Exceeds",
                 },
             });
         });

__tests__/unit/crypto/transactions/transaction.test.ts

@@ -1,10 +1,12 @@
 import "jest-extended";
 
+import { Address } from "../../../../packages/crypto/src/identities";
 import { configManager } from "../../../../packages/crypto/src/managers";
 import { devnet } from "../../../../packages/crypto/src/networks";
 import { TransactionFactory } from "../../../../packages/crypto/src/transactions";
-import { BigNumber } from "../../../../packages/crypto/src/utils";
 import { Two } from "../../../../packages/crypto/src/transactions/types";
+import { BuilderFactory } from "../../../../packages/crypto/src/transactions";
+import { BigNumber } from "../../../../packages/crypto/src/utils";
 
 describe("Transaction", () => {
     describe("should deserialize correctly some tests transactions", () => {
@@ -176,4 +178,37 @@ describe("Transaction", () => {
             devnet.milestones.pop();
         });
     });
+
+    describe("toString", () => {
+        it("should describe v1 transaction", () => {
+            configManager.getMilestone().aip11 = false;
+
+            const senderAddress = Address.fromPassphrase("sender's secret");
+            const recipientAddress = Address.fromPassphrase("recipient's secret");
+            const transaction = BuilderFactory.transfer()
+                .version(1)
+                .amount("100")
+                .recipientId(recipientAddress)
+                .sign("sender's secret")
+                .build();
+
+            expect(String(transaction)).toMatch(new RegExp(`^${senderAddress} transfer v1 [0-9a-f]{64}$`));
+        });
+
+        it("should describe v2 transaction", () => {
+            configManager.getMilestone().aip11 = true;
+
+            const senderAddress = Address.fromPassphrase("sender's secret");
+            const recipientAddress = Address.fromPassphrase("recipient's secret");
+            const transaction = BuilderFactory.transfer()
+                .version(2)
+                .amount("100")
+                .recipientId(recipientAddress)
+                .nonce("1")
+                .sign("sender's secret")
+                .build();
+
+            expect(String(transaction)).toMatch(new RegExp(`^${senderAddress} #1 transfer v2 [0-9a-f]{64}$`));
+        });
+    });
 });

packages/core-api/src/controllers/transactions.ts

@@ -1,12 +1,12 @@
 import { Models, Repositories } from "@arkecosystem/core-database";
 import { Container, Contracts, Utils as AppUtils } from "@arkecosystem/core-kernel";
 import { Handlers } from "@arkecosystem/core-transactions";
-import { Interfaces } from "@arkecosystem/crypto";
 import Boom from "@hapi/boom";
 import Hapi from "@hapi/hapi";
 
 import { TransactionResource } from "../resources";
 import { Controller } from "./controller";
+import { Interfaces } from "@arkecosystem/crypto";
 
 @Container.injectable()
 export class TransactionsController extends Controller {
@@ -16,8 +16,8 @@ export class TransactionsController extends Controller {
     @Container.inject(Container.Identifiers.BlockchainService)
     protected readonly blockchain!: Contracts.Blockchain.Blockchain;
 
-    @Container.inject(Container.Identifiers.TransactionPoolService)
-    private readonly transactionPool!: Contracts.TransactionPool.Connection;
+    @Container.inject(Container.Identifiers.TransactionPoolQuery)
+    private readonly poolQuery!: Contracts.TransactionPool.Query;
 
     @Container.inject(Container.Identifiers.TransactionRepository)
     private readonly transactionRepository!: Repositories.TransactionRepository;
@@ -63,30 +63,30 @@ export class TransactionsController extends Controller {
     }
 
     public async unconfirmed(request: Hapi.Request, h: Hapi.ResponseToolkit) {
-        const pagination = super.paginate(request);
-        const transactions = await this.transactionPool.getTransactions(pagination.offset, pagination.limit);
-        const poolSize = await this.transactionPool.getPoolSize();
-        const data = transactions.map(t => ({ serialized: t.serialized.toString("hex") }));
-
-        return super.toPagination(
-            { count: poolSize, rows: data },
-            TransactionResource,
-            (request.query.transform as unknown) as boolean,
+        const pagination: Repositories.Search.SearchPagination = super.paginate(request);
+        const all: Interfaces.ITransaction[] = Array.from(this.poolQuery.getAllFromHighestPriority());
+        const transactions: Interfaces.ITransaction[] = all.slice(
+            pagination.offset,
+            pagination.offset + pagination.limit,
         );
+        const rows = transactions.map(t => ({ serialized: t.serialized.toString("hex") }));
+
+        return super.toPagination({ count: all.length, rows }, TransactionResource, !!request.query.transform);
     }
 
     public async showUnconfirmed(request: Hapi.Request, h: Hapi.ResponseToolkit) {
-        const transaction: Interfaces.ITransaction | undefined = await this.transactionPool.getTransaction(
-            request.params.id,
-        );
+        const transactionQuery: Contracts.TransactionPool.QueryIterable = this.poolQuery
+            .getAllFromHighestPriority()
+            .whereId(request.params.id);
 
-        if (!transaction) {
+        if (transactionQuery.has() === false) {
             return Boom.notFound("Transaction not found");
         }
 
+        const transaction: Interfaces.ITransaction = transactionQuery.first();
         const data = { id: transaction.id, serialized: transaction.serialized.toString("hex") };
 
-        return super.respondWithResource(data, TransactionResource, (request.query.transform as unknown) as boolean);
+        return super.respondWithResource(data, TransactionResource, !!request.query.transform);
     }
 
     public async search(request: Hapi.Request, h: Hapi.ResponseToolkit) {

packages/core-blockchain/src/blockchain.ts

@@ -33,7 +33,7 @@ export class Blockchain implements Contracts.Blockchain.Blockchain {
     private readonly blockRepository!: Repositories.BlockRepository;
 
     @Container.inject(Container.Identifiers.TransactionPoolService)
-    private readonly transactionPool!: Contracts.TransactionPool.Connection;
+    private readonly transactionPool!: Contracts.TransactionPool.Service;
 
     @Container.inject(Container.Identifiers.StateMachine)
     private readonly stateMachine!: StateMachine;
@@ -330,7 +330,7 @@ export class Blockchain implements Contracts.Blockchain.Blockchain {
         await this.blockRepository.deleteBlocks(removedBlocks);
 
         if (this.transactionPool) {
-            await this.transactionPool.replay(removedTransactions.reverse());
+            this.transactionPool.readdTransactions(removedTransactions.reverse());
         }
     }
 

packages/core-blockchain/src/processor/block-processor.ts

@@ -35,9 +35,6 @@ export class BlockProcessor {
     @Container.inject(Container.Identifiers.TransactionRepository)
     private readonly transactionRepository!: Repositories.TransactionRepository;
 
-    @Container.inject(Container.Identifiers.TransactionPoolService)
-    private readonly transactionPool!: Contracts.TransactionPool.Connection;
-
     public async process(block: Interfaces.IBlock): Promise<BlockProcessorResult> {
         if (Utils.isException(block.data.id)) {
             return this.app.resolve<ExceptionHandler>(ExceptionHandler).execute(block);
@@ -123,10 +120,6 @@ export class BlockProcessor {
             );
 
             if (forgedIds.length > 0) {
-                if (this.transactionPool) {
-                    this.transactionPool.removeTransactionsById(forgedIds);
-                }
-
                 this.logger.warning(
                     `Block ${block.data.height.toLocaleString()} disregarded, because it contains already forged transactions`,
                 );