Use subscription at rejectIfBlockTimeout when the provider supports subscription

CHANGELOG.md

@@ -739,6 +739,7 @@ should use 4.0.1-alpha.0 for testing.
 #### web3-core
 
 -   If the response error was `execution reverted`, raise `ContractExecutionError` and pass the response error to it in order to be set as `innerError` (this innerError will be decoded at web3-eth-contract if its ABI was provided according to EIP-838). (#5434)
+-   Added a new configuration variable `enableExperimentalFeatures`. (#5481)
 -   `registerPlugin` method to `Web3Context` (#5393)
 -   `Web3PluginBase` exported abstract class (#5393)
 -   `Web3EthPluginBase` exported abstract class (#5393)
@@ -789,6 +790,11 @@ should use 4.0.1-alpha.0 for testing.
 
 -   Moved `SignerError` from `web3-errors/src/errors/signature_errors.ts` to `web3-errors/src/errors/transaction_errors.ts`, and renamed it to `TransactionSigningError` (#5462)
 
+#### web3-eth
+
+-   Use subscription at `rejectIfBlockTimeout` when the provider supports subscription. Implement this as an experimental feature (if `useSubscriptionWhenCheckingBlockTimeout` at `enableExperimentalFeatures` is `true`). (#5481)
+-   At some test cases, optimize some codes. (#5481)
+
 ### Fixed
 
 #### web3-error

fixtures/utils.ts

@@ -15,11 +15,14 @@ You should have received a copy of the GNU Lesser General Public License
 along with web3.js.  If not, see <http://www.gnu.org/licenses/>.
 */
 export const processAsync = async (
-	processFunc: (resolver: (value: unknown) => void) => Promise<unknown> | unknown,
+	processFunc: (
+		resolver: (value: unknown) => void,
+		reject: (value: unknown) => void,
+	) => Promise<unknown> | unknown,
 ) =>
-	new Promise(resolve => {
+	new Promise((resolve, reject) => {
 		(async () => {
-			await processFunc(resolve);
+			await processFunc(resolve, reject);
 		})() as unknown;
 	});
 

packages/web3-core/CHANGELOG.md

@@ -50,3 +50,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 -   Default value for `API` generic for `Web3ContextInitOptions` from `any` to `unknown` (#5393)
 -   Added validation when `defaultHardfork` and `defaultCommon.hardfork` are different in web3config
 -   Added validation when `defaultChain` and `defaultCommon.basechain` are different in web3config
+-   Added a new configuration variable `enableExperimentalFeatures`. (#5481)
+

packages/web3-core/src/web3_config.ts

@@ -44,6 +44,10 @@ export interface Web3ConfigOptions {
 	defaultCommon?: Common;
 	defaultTransactionType: Numbers;
 	defaultMaxPriorityFeePerGas: Numbers;
+	enableExperimentalFeatures: {
+		useSubscriptionWhenCheckingBlockTimeout: boolean;
+		// other experimental features...
+	};
 	transactionBuilder?: TransactionBuilder;
 	transactionTypeParser?: TransactionTypeParser;
 }
@@ -80,6 +84,9 @@ export abstract class Web3Config
 		defaultCommon: undefined,
 		defaultTransactionType: '0x0',
 		defaultMaxPriorityFeePerGas: toHex(2500000000),
+		enableExperimentalFeatures: {
+			useSubscriptionWhenCheckingBlockTimeout: false,
+		},
 		transactionBuilder: undefined,
 		transactionTypeParser: undefined,
 	};
@@ -331,6 +338,23 @@ export abstract class Web3Config
 		this._config.blockHeaderTimeout = val;
 	}
 
+	/**
+	 * The enableExperimentalFeatures is used to enable trying new experimental features that are still not fully implemented or not fully tested or still have some related issues.
+	 * Default is `false` for every feature.
+	 */
+	public get enableExperimentalFeatures() {
+		return this._config.enableExperimentalFeatures;
+	}
+
+	/**
+	 * Will set the enableExperimentalFeatures
+	 */
+	public set enableExperimentalFeatures(val) {
+		this._triggerConfigChange('enableExperimentalFeatures', val);
+
+		this._config.enableExperimentalFeatures = val;
+	}
+
 	public get maxListenersWarningThreshold() {
 		return this._config.maxListenersWarningThreshold;
 	}

packages/web3-core/src/web3_subscription_manager.ts

@@ -99,7 +99,9 @@ export class Web3SubscriptionManager<
 
 	public async removeSubscription(sub: InstanceType<RegisteredSubs[keyof RegisteredSubs]>) {
 		if (isNullish(sub.id)) {
-			throw new SubscriptionError('Subscription is not subscribed yet.');
+			throw new SubscriptionError(
+				'Subscription is not subscribed yet. Or, had already been unsubscribed but not through the Subscription Manager.',
+			);
 		}
 
 		if (!this._subscriptions.has(sub.id)) {

packages/web3-core/test/unit/__snapshots__/web3_context.test.ts.snap

@@ -13,6 +13,9 @@ Object {
     "defaultMaxPriorityFeePerGas": "0x9502f900",
     "defaultNetworkId": undefined,
     "defaultTransactionType": "0x0",
+    "enableExperimentalFeatures": Object {
+      "useSubscriptionWhenCheckingBlockTimeout": false,
+    },
     "handleRevert": false,
     "maxListenersWarningThreshold": 100,
     "transactionBlockTimeout": 50,

packages/web3-core/test/unit/web3_config.test.ts

@@ -28,6 +28,9 @@ const defaultConfig = {
 	defaultNetworkId: undefined,
 	defaultCommon: undefined,
 	defaultHardfork: 'london',
+	enableExperimentalFeatures: {
+		useSubscriptionWhenCheckingBlockTimeout: false,
+	},
 	handleRevert: false,
 	maxListenersWarningThreshold: 100,
 	transactionBlockTimeout: 50,

packages/web3-eth-contract/src/encoding.ts

@@ -167,7 +167,10 @@ export const decodeEventABI = (
 		...result,
 		returnValues: decodeLog([...(modifiedEvent.inputs ?? [])], data.data, argTopics),
 		event: modifiedEvent.name,
-		signature: modifiedEvent.anonymous || !data.topics[0] ? undefined : data.topics[0],
+		signature:
+			modifiedEvent.anonymous || !data.topics || data.topics.length === 0 || !data.topics[0]
+				? undefined
+				: data.topics[0],
 
 		raw: {
 			data: data.data,

packages/web3-eth-contract/test/integration/contract_events.test.ts

@@ -55,10 +55,11 @@ describe('contract', () => {
 		itIf(isWs)('should trigger the "contract.events.<eventName>"', async () => {
 			// eslint-disable-next-line jest/no-standalone-expect
 			return expect(
-				processAsync(async resolve => {
+				processAsync(async (resolve, reject) => {
 					const event = contractDeployed.events.MultiValueEvent();
 
 					event.on('data', resolve);
+					event.on('error', reject);
 
 					// trigger event
 					await contractDeployed.methods
@@ -75,12 +76,13 @@ describe('contract', () => {
 		itIf(isWs)(
 			'should trigger the "contract.events.<eventName>" for indexed parameters',
 			async () => {
-				const res = await processAsync(async resolve => {
+				const res = await processAsync(async (resolve, reject) => {
 					const event = contractDeployed.events.MultiValueIndexedEvent({
 						filter: { val: 100 },
 					});
 
 					event.on('data', resolve);
+					event.on('error', reject);
 
 					// trigger event
 					await contractDeployed.methods
@@ -102,12 +104,13 @@ describe('contract', () => {
 			async () => {
 				// eslint-disable-next-line jest/no-standalone-expect
 				return expect(
-					processAsync(async resolve => {
+					processAsync(async (resolve, reject) => {
 						const event = contractDeployed.events.MultiValueEvent({
 							fromBlock: 'latest',
 						});
 
 						event.on('data', resolve);
+						event.on('error', reject);
 
 						// trigger event
 						await contractDeployed.methods
@@ -121,7 +124,9 @@ describe('contract', () => {
 				);
 			},
 		);
+	});
 
+	describe('events subscription with HTTP', () => {
 		itIf(isHttp)('should fail to subscribe', async () => {
 			// eslint-disable-next-line no-async-promise-executor, @typescript-eslint/no-misused-promises
 			const failedSubscriptionPromise = new Promise<void>((resolve, reject) => {

packages/web3-eth/CHANGELOG.md

@@ -50,6 +50,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 -   Moved `rpc_methods` tests to `web3-rpc-methods` (#5441)
 -   [setimmediate](https://github.com/yuzujs/setImmediate) package to polyfill [setImmediate](https://nodejs.org/api/timers.html#setimmediatecallback-args) for browsers (#5450)
 -   Implemented the logic for `transactionBlockTimeout` (#5294)
+-   Use subscription at `rejectIfBlockTimeout` when the provider supports subscription. Implement this as an experimental feature (if `useSubscriptionWhenCheckingBlockTimeout` at `enableExperimentalFeatures` is `true`). (#5481)
+-   At some test cases, optimize some codes. (#5481)
 
 ### Removed
 

packages/web3-eth/src/utils/reject_if_block_timeout.ts

@@ -14,40 +14,164 @@ GNU Lesser General Public License for more details.
 You should have received a copy of the GNU Lesser General Public License
 along with web3.js.  If not, see <http://www.gnu.org/licenses/>.
 */
-import { EthExecutionAPI, Bytes } from 'web3-types';
+import { EthExecutionAPI, Bytes, Web3BaseProvider, BlockHeaderOutput } from 'web3-types';
 import { Web3Context } from 'web3-core';
 import { rejectIfConditionAtInterval } from 'web3-utils';
 
 import { TransactionBlockTimeoutError } from 'web3-errors';
 import { NUMBER_DATA_FORMAT } from '../constants';
 // eslint-disable-next-line import/no-cycle
 import { getBlockNumber } from '../rpc_method_wrappers';
+import { NewHeadsSubscription } from '../web3_subscriptions';
 
-/* TODO: After merge, there will be constant block mining time (exactly 12 second each block, except slot missed that currently happens in <1% of slots. ) so we can optimize following function
-for POS NWs, we can skip checking getBlockNumber(); after interval and calculate only based on time  that certain num of blocked are mined after that for internal double check, can do one getBlockNumber() call and timeout. 
-*/
-export function rejectIfBlockTimeout(
+export interface ResourceCleaner {
+	clean: () => void;
+}
+
+function resolveByPolling(
 	web3Context: Web3Context<EthExecutionAPI>,
 	starterBlockNumber: number,
-	interval: number,
 	transactionHash?: Bytes,
-): [NodeJS.Timer, Promise<never>] {
-	return rejectIfConditionAtInterval(async () => {
-		let lastBlockNumber;
-		try {
-			lastBlockNumber = await getBlockNumber(web3Context, NUMBER_DATA_FORMAT);
-		} catch (error) {
-			console.warn('An error happen while trying to get the block number', error);
+): [Promise<never>, ResourceCleaner] {
+	const pollingInterval = web3Context.transactionPollingInterval;
+	const [intervalId, promiseToError]: [NodeJS.Timer, Promise<never>] =
+		rejectIfConditionAtInterval(async () => {
+			let lastBlockNumber;
+			try {
+				lastBlockNumber = await getBlockNumber(web3Context, NUMBER_DATA_FORMAT);
+			} catch (error) {
+				console.warn('An error happen while trying to get the block number', error);
+				return undefined;
+			}
+			const numberOfBlocks = lastBlockNumber - starterBlockNumber;
+			if (numberOfBlocks >= web3Context.transactionBlockTimeout) {
+				return new TransactionBlockTimeoutError({
+					starterBlockNumber,
+					numberOfBlocks,
+					transactionHash,
+				});
+			}
 			return undefined;
+		}, pollingInterval);
+
+	const clean = () => {
+		clearInterval(intervalId);
+	};
+
+	return [promiseToError, { clean }];
+}
+
+async function resolveBySubscription(
+	web3Context: Web3Context<EthExecutionAPI>,
+	starterBlockNumber: number,
+	transactionHash?: Bytes,
+): Promise<[Promise<never>, ResourceCleaner]> {
+	// The following variable will stay true except if the data arrived,
+	//	or if watching started after an error had occurred.
+	let needToWatchLater = true;
+
+	let subscription: NewHeadsSubscription;
+	let resourceCleaner: ResourceCleaner;
+	// internal helper function
+	function revertToPolling(
+		reject: (value: Error | PromiseLike<Error>) => void,
+		previousError?: Error,
+	) {
+		if (previousError) {
+			console.warn('error happened at subscription. So revert to polling...', previousError);
 		}
-		const numberOfBlocks = lastBlockNumber - starterBlockNumber;
-		if (numberOfBlocks >= web3Context.transactionBlockTimeout) {
-			return new TransactionBlockTimeoutError({
-				starterBlockNumber,
-				numberOfBlocks,
-				transactionHash,
+		resourceCleaner.clean();
+
+		needToWatchLater = false;
+		const [promiseToError, newResourceCleaner] = resolveByPolling(
+			web3Context,
+			starterBlockNumber,
+			transactionHash,
+		);
+		resourceCleaner.clean = newResourceCleaner.clean;
+		promiseToError.catch(error => reject(error as Error));
+	}
+	try {
+		subscription = (await web3Context.subscriptionManager?.subscribe(
+			'newHeads',
+		)) as unknown as NewHeadsSubscription;
+		resourceCleaner = {
+			clean: () => {
+				// Remove the subscription, if it was not removed somewhere
+				// 	else by calling, for example, subscriptionManager.clear()
+				if (subscription.id) {
+					web3Context.subscriptionManager
+						?.removeSubscription(subscription)
+						.then(() => {
+							// Subscription ended successfully
+						})
+						.catch(() => {
+							// An error happened while ending subscription. But no need to take any action.
+						});
+				}
+			},
+		};
+	} catch (error) {
+		return resolveByPolling(web3Context, starterBlockNumber, transactionHash);
+	}
+	const promiseToError: Promise<never> = new Promise((_, reject) => {
+		try {
+			subscription.on('data', (lastBlockHeader: BlockHeaderOutput) => {
+				needToWatchLater = false;
+				if (!lastBlockHeader?.number) {
+					return;
+				}
+				const numberOfBlocks = Number(
+					BigInt(lastBlockHeader.number) - BigInt(starterBlockNumber),
+				);
+
+				if (numberOfBlocks >= web3Context.transactionBlockTimeout) {
+					// Transaction Block Timeout is known to be reached by subscribing to new heads
+					reject(
+						new TransactionBlockTimeoutError({
+							starterBlockNumber,
+							numberOfBlocks,
+							transactionHash,
+						}),
+					);
+				}
+			});
+			subscription.on('error', error => {
+				revertToPolling(reject, error);
 			});
+		} catch (error) {
+			revertToPolling(reject, error as Error);
 		}
-		return undefined;
-	}, interval);
+
+		// Fallback to polling if tx receipt didn't arrived in "blockHeaderTimeout" [10 seconds]
+		setTimeout(() => {
+			if (needToWatchLater) {
+				revertToPolling(reject);
+			}
+		}, web3Context.blockHeaderTimeout * 1000);
+	});
+
+	return [promiseToError, resourceCleaner];
+}
+
+/* TODO: After merge, there will be constant block mining time (exactly 12 second each block, except slot missed that currently happens in <1% of slots. ) so we can optimize following function
+for POS NWs, we can skip checking getBlockNumber(); after interval and calculate only based on time  that certain num of blocked are mined after that for internal double check, can do one getBlockNumber() call and timeout. 
+*/
+export async function rejectIfBlockTimeout(
+	web3Context: Web3Context<EthExecutionAPI>,
+	transactionHash?: Bytes,
+): Promise<[Promise<never>, ResourceCleaner]> {
+	const provider: Web3BaseProvider = web3Context.requestManager.provider as Web3BaseProvider;
+	let callingRes: [Promise<never>, ResourceCleaner];
+	const starterBlockNumber = await getBlockNumber(web3Context, NUMBER_DATA_FORMAT);
+	// TODO: once https://github.com/web3/web3.js/issues/5521 is implemented, remove checking for `enableExperimentalFeatures.useSubscriptionWhenCheckingBlockTimeout`
+	if (
+		provider.supportsSubscriptions() &&
+		web3Context.enableExperimentalFeatures.useSubscriptionWhenCheckingBlockTimeout
+	) {
+		callingRes = await resolveBySubscription(web3Context, starterBlockNumber, transactionHash);
+	} else {
+		callingRes = resolveByPolling(web3Context, starterBlockNumber, transactionHash);
+	}
+	return callingRes;
 }