feat(transaction-pool-service): fetch transactions for forging in batches (#1296)

* Implement transaction selector

* Add selector

* Register selector

* Update worker

* Clean packages on ts:token

* Update forger

* Pass transaction data

* Use block-round identifier

* Use iterator

* Inline

* Use initializer

* Reduce complexity

* Improve code

* Improve code

* Handle gasLeft < 0

* Gas used as number

* Remove identifier

* Clear selector on commit

* Cleanup worker

* Fix consensus tests

* Fix workers

* Cleanup worker

* Fix pool workers

* style: resolve style guide violations [ci-lint-fix]

* Fix validator unit tests

* Fix functional tests

Author: sebastijankuzner

package.json

@@ -12,7 +12,7 @@
 		"build:rs": "lerna run build-rs --scope=@mainsail/evm",
 		"build:rs:token": "echo \"force-rs $(date +%s)\" > .build-rs",
 		"build:ts": "lerna run build",
-		"build:ts:token": "echo \"force-ts $(date +%s)\" > .build-ts",
+		"build:ts:token": "echo \"force-ts $(date +%s)\" > .build-ts && pnpm run clean:packages",
 		"clean": "pnpm run clean:packages && rm -r .nx",
 		"clean:all": "pnpm run clean && pnpm run clean:node_modules",
 		"clean:node_modules": "rm -rf packages/*/node_modules && rm -rf node_modules",

packages/constants/source/identifiers.ts

@@ -175,7 +175,6 @@ export const Identifiers = {
 	},
 	Forger: {
 		Block: Symbol("Forger<Block>"),
-		Transaction: Symbol("Forger<Transaction>"),
 	},
 	P2P: {
 		ApiNode: {
@@ -301,6 +300,7 @@ export const Identifiers = {
 		Processor: Symbol("TransactionPool<Processor>"),
 		ProcessorExtension: Symbol("TransactionPool<ProcessorExtension>"),
 		Query: Symbol("TransactionPool<Query>"),
+		Selector: Symbol("TransactionPool<Selector>"),
 		SenderMempool: {
 			Factory: Symbol("TransactionPool<SenderMempool.Factory>"),
 		},

packages/contracts/source/contracts/transaction-pool/index.ts

@@ -3,6 +3,7 @@ export * from "./client.js";
 export * from "./mempool.js";
 export * from "./processor.js";
 export * from "./query.js";
+export * from "./selector.js";
 export * from "./sender-mempool.js";
 export * from "./sender-state.js";
 export * from "./service.js";

packages/contracts/source/contracts/transaction-pool/selector.ts

@@ -0,0 +1,17 @@
+import type { TransactionData } from "../crypto/index.js";
+
+export type GetBatchOptions = {
+	blockRound: string;
+	maxSize: number;
+	maxBytes: number;
+};
+
+export type GetBatchResult = {
+	transactions: TransactionData[];
+	remaining: number;
+};
+
+export interface Selector {
+	getBatch(options: GetBatchOptions): Promise<GetBatchResult>;
+	clear(): void;
+}

packages/contracts/source/contracts/transaction-pool/worker.ts

@@ -2,12 +2,13 @@ import type { CommitHandler } from "../crypto/index.js";
 import type { EventListener } from "../kernel/index.js";
 import type { EventCallback, Subprocess } from "../kernel/ipc.js";
 import type { KeyValuePair } from "../types/index.js";
+import type { GetBatchResult, GetBatchOptions } from "./selector.js";
 
 export type WorkerFlags = KeyValuePair;
 
 export interface WorkerScriptHandler {
 	boot(flags: WorkerFlags): Promise<void>;
-	getTransactions(): Promise<string[]>;
+	getTransactions(options: GetBatchOptions): Promise<GetBatchResult>;
 	removeTransaction(address: string, id: string): Promise<void>;
 	commit(height: number, sendersAddresses: string[], consumedGas: number, isSyncing: boolean): Promise<void>;
 	setPeer(ip: string): Promise<void>;
@@ -22,9 +23,8 @@ export type WorkerSubprocess = Subprocess<WorkerScriptHandler>;
 
 export type WorkerSubprocessFactory = () => WorkerSubprocess;
 
-export interface Worker extends Omit<WorkerScriptHandler, "commit" | "getTransactions">, CommitHandler, EventListener {
+export interface Worker extends Omit<WorkerScriptHandler, "commit">, CommitHandler, EventListener {
 	getQueueSize(): number;
 	kill(): Promise<number>;
-	getTransactionBytes(): Promise<Buffer[]>;
 	registerEventHandler<T>(event: string, callback: EventCallback<T>): void;
 }

packages/forger/source/block-forger.ts

@@ -4,8 +4,13 @@ import { Identifiers } from "@mainsail/constants";
 import { inject, injectable } from "@mainsail/container";
 import { assert } from "@mainsail/utils";
 
+import { TransactionForger } from "./transaction-forger.js";
+
 @injectable()
 export class BlockForger implements Contracts.Forger.BlockForger {
+	@inject(Identifiers.Application.Instance)
+	private readonly app!: Contracts.Kernel.Application;
+
 	@inject(Identifiers.Cryptography.Configuration)
 	private readonly cryptoConfiguration!: Contracts.Crypto.Configuration;
 
@@ -18,9 +23,6 @@ export class BlockForger implements Contracts.Forger.BlockForger {
 	@inject(Identifiers.Cryptography.Hash.Factory)
 	private readonly hashFactory!: Contracts.Crypto.HashFactory;
 
-	@inject(Identifiers.Forger.Transaction)
-	protected readonly transactionForger!: Contracts.Forger.TransactionForger;
-
 	@inject(Identifiers.BlockchainUtils.FeeCalculator)
 	protected readonly gasFeeCalculator!: Contracts.BlockchainUtils.FeeCalculator;
 
@@ -32,14 +34,12 @@ export class BlockForger implements Contracts.Forger.BlockForger {
 		const previousBlock = this.stateStore.getLastBlock();
 		const blockNumber = previousBlock.number + 1;
 
-		const { fee, gasUsed, logsBloom, stateRoot, transactions } = await this.transactionForger.getTransactions(
-			generatorAddress,
-			timestamp,
-			{
-				blockNumber: BigInt(blockNumber),
-				round: BigInt(round),
-			},
-		);
+		const transactionForger = this.app.resolve(TransactionForger).initialize(generatorAddress, timestamp, {
+			blockNumber: BigInt(blockNumber),
+			round: BigInt(round),
+		});
+
+		const { fee, gasUsed, logsBloom, stateRoot, transactions } = await transactionForger.getTransactions();
 		return this.#makeBlock(round, generatorAddress, logsBloom, stateRoot, transactions, timestamp, gasUsed, fee);
 	}
 

packages/forger/source/service-provider.ts

@@ -4,12 +4,10 @@ import { Providers } from "@mainsail/kernel";
 import Joi from "joi";
 
 import { BlockForger } from "./block-forger.js";
-import { TransactionForger } from "./transaction-forger.js";
 
 @injectable()
 export class ServiceProvider extends Providers.ServiceProvider {
 	public async register(): Promise<void> {
-		this.app.bind(Identifiers.Forger.Transaction).to(TransactionForger).inSingletonScope();
 		this.app.bind(Identifiers.Forger.Block).to(BlockForger).inSingletonScope();
 	}