try rebroadcast on commit

Author: oXtxNt9U

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

@@ -5,6 +5,6 @@ export interface Service {
 
 	addTransaction(transaction: Transaction): Promise<void>;
 	reAddTransactions(): Promise<void>;
-	commit(sendersAddresses: string[]): Promise<void>;
+	commit(sendersAddresses: string[], consumedGas: number): Promise<void>;
 	flush(): Promise<void>;
 }

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

@@ -9,7 +9,7 @@ export interface WorkerScriptHandler {
 	boot(flags: WorkerFlags): Promise<void>;
 	getTransactions(): Promise<string[]>;
 	removeTransaction(address: string, id: string): Promise<void>;
-	commit(height: number, sendersAddresses: string[]): Promise<void>;
+	commit(height: number, sendersAddresses: string[], consumedGas: number): Promise<void>;
 	setPeer(ip: string): Promise<void>;
 	forgetPeer(ip: string): Promise<void>;
 	start(height: number): Promise<void>;

packages/transaction-pool-service/source/service.ts

@@ -16,6 +16,12 @@ export class Service implements Contracts.TransactionPool.Service {
 	@inject(Identifiers.State.Store)
 	private readonly stateStore!: Contracts.State.Store;
 
+	@inject(Identifiers.TransactionPool.Broadcaster)
+	private readonly broadcaster!: Contracts.TransactionPool.Broadcaster;
+
+	@inject(Identifiers.Cryptography.Configuration)
+	private readonly cryptoConfiguration!: Contracts.Crypto.Configuration;
+
 	@inject(Identifiers.TransactionPool.Storage)
 	private readonly storage!: Contracts.TransactionPool.Storage;
 
@@ -55,7 +61,7 @@ export class Service implements Contracts.TransactionPool.Service {
 		return this.mempool.getSize();
 	}
 
-	public async commit(sendersAddresses: string[]): Promise<void> {
+	public async commit(sendersAddresses: string[], consumedGas: number): Promise<void> {
 		await this.#lock.runExclusive(async () => {
 			if (this.#disposed) {
 				return;
@@ -70,6 +76,8 @@ export class Service implements Contracts.TransactionPool.Service {
 			}
 
 			await this.#cleanUp();
+
+			await this.#rebroadcastStorageTransactions(consumedGas);
 		});
 	}
 
@@ -242,4 +250,35 @@ export class Service implements Contracts.TransactionPool.Service {
 
 		await this.mempool.addTransaction(transaction);
 	}
+
+	async #rebroadcastStorageTransactions(consumedGas: number): Promise<void> {
+		const blockNumber = this.stateStore.getBlockNumber();
+		const milestones = this.cryptoConfiguration.getMilestone(blockNumber);
+
+		const threshold = this.pluginConfiguration.getRequired<number>(
+			"maxBlockGasUtilizationTransactionRebroadcastThreshold",
+		);
+
+		// If block is not full rebroadcast local transactions.
+		if (consumedGas > milestones.block.maxGasLimit * (threshold / 100)) {
+			return;
+		}
+
+		const limit = this.pluginConfiguration.getRequired<number>("maxTransactionsPerRequest");
+		const broadcastTransactions: Contracts.Crypto.Transaction[] = [];
+
+		// Get old transactions up to current block number.
+		for (const { serialized } of this.storage.getOldTransactions(blockNumber, limit)) {
+			const transaction = await this.transactionFactory.fromBytes(serialized);
+			broadcastTransactions.push(transaction);
+		}
+
+		if (broadcastTransactions.length > 0) {
+			this.logger.info(`Rebroadcasting ${broadcastTransactions.length} transaction(s) from storage`);
+
+			this.broadcaster
+				.broadcastTransactions(broadcastTransactions)
+				.catch((error) => this.logger.error(error.stack));
+		}
+	}
 }

packages/transaction-pool-worker/source/handlers/commit.ts

@@ -15,14 +15,14 @@ export class CommitHandler {
 	@inject(Identifiers.Services.Log.Service)
 	protected readonly logger!: Contracts.Kernel.Logger;
 
-	public async handle(blockNumber: number, sendersAddresses: string[]): Promise<void> {
+	public async handle(blockNumber: number, sendersAddresses: string[], consumedGas: number): Promise<void> {
 		try {
 			this.stateStore.setBlockNumber(blockNumber);
 
 			if (this.configuration.isNewMilestone()) {
 				void this.transactionPoolService.reAddTransactions();
 			} else {
-				await this.transactionPoolService.commit(sendersAddresses);
+				await this.transactionPoolService.commit(sendersAddresses, consumedGas);
 			}
 		} catch (error) {
 			throw new Error(`Failed to commit block: ${error.message}`);

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

@@ -31,8 +31,8 @@ export class WorkerScriptHandler implements Contracts.TransactionPool.WorkerScri
 		await this.#app.resolve(StartHandler).handle(height);
 	}
 
-	public async commit(height: number, sendersAddresses: string[]): Promise<void> {
-		await this.#app.resolve(CommitHandler).handle(height, sendersAddresses);
+	public async commit(height: number, sendersAddresses: string[], consumedGas: number): Promise<void> {
+		await this.#app.resolve(CommitHandler).handle(height, sendersAddresses, consumedGas);
 	}
 
 	public async getTransactions(): Promise<string[]> {

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

@@ -50,11 +50,17 @@ export class Worker implements Contracts.TransactionPool.Worker {
 	async onCommit(unit: Contracts.Processor.ProcessableUnit): Promise<void> {
 		const sendersAddresses: Set<string> = new Set();
 
-		for (const transaction of unit.getBlock().transactions) {
+		const block = unit.getBlock();
+		for (const transaction of block.transactions) {
 			sendersAddresses.add(transaction.data.from);
 		}
 
-		await this.ipcSubprocess.sendRequest("commit", unit.blockNumber, [...sendersAddresses.keys()]);
+		await this.ipcSubprocess.sendRequest(
+			"commit",
+			unit.blockNumber,
+			[...sendersAddresses.keys()],
+			block.header.gasUsed,
+		);
 	}
 
 	public async start(blockNumber: number): Promise<void> {