CCU creation fails sometimes when the node fails to save data on start of node

framework-plugins/lisk-framework-chain-connector-plugin/src/chain_connector_plugin.ts

@@ -60,7 +60,6 @@ import { Endpoint } from './endpoint';
 import { configSchema } from './schemas';
 import {
 	ChainConnectorPluginConfig,
-	SentCCUs,
 	BlockHeader,
 	ProveResponseJSON,
 	BFTParametersJSON,
@@ -106,8 +105,8 @@ export class ChainConnectorPlugin extends BasePlugin<ChainConnectorPluginConfig>
 	private _receivingChainID!: Buffer;
 	private _isReceivingChainMainchain!: boolean;
 	private _registrationHeight!: number;
-	private readonly _sentCCUs: SentCCUs = [];
 	private _privateKey!: Buffer;
+	private _ccuSaveLimit!: number;
 
 	public get nodeModulePath(): string {
 		return __filename;
@@ -124,6 +123,7 @@ export class ChainConnectorPlugin extends BasePlugin<ChainConnectorPluginConfig>
 		this._maxCCUSize = this.config.maxCCUSize;
 		this._isSaveCCU = this.config.isSaveCCU;
 		this._registrationHeight = this.config.registrationHeight;
+		this._ccuSaveLimit = this.config.ccuSaveLimit;
 		const { password, encryptedPrivateKey } = this.config;
 		if (password) {
 			const parsedEncryptedKey = encrypt.parseEncryptedMessage(encryptedPrivateKey);
@@ -139,8 +139,6 @@ export class ChainConnectorPlugin extends BasePlugin<ChainConnectorPluginConfig>
 		this._chainConnectorStore = new ChainConnectorStore(this._chainConnectorPluginDB);
 		this.endpoint.load(this._chainConnectorStore);
 
-		await this._initializeReceivingChainClient();
-
 		this._sendingChainClient = this.apiClient;
 
 		this._ownChainID = Buffer.from(
@@ -169,7 +167,7 @@ export class ChainConnectorPlugin extends BasePlugin<ChainConnectorPluginConfig>
 
 	public async unload(): Promise<void> {
 		if (this._receivingChainClient) {
-			await this._sendingChainClient.disconnect();
+			await this._receivingChainClient.disconnect();
 		}
 		if (this._sendingChainClient) {
 			await this._sendingChainClient.disconnect();
@@ -682,6 +680,16 @@ export class ChainConnectorPlugin extends BasePlugin<ChainConnectorPluginConfig>
 					validatorsData.certificateThreshold >= BigInt(this._lastCertificate.height),
 			),
 		);
+		// Delete CCUs
+		// When given -1 then there is no limit
+		if (this._ccuSaveLimit !== -1) {
+			const listOfCCUs = await this._chainConnectorStore.getListOfCCUs();
+			if (listOfCCUs.length > this._ccuSaveLimit) {
+				await this._chainConnectorStore.setListOfCCUs(
+					listOfCCUs.slice(0, listOfCCUs.length - this._ccuSaveLimit),
+				);
+			}
+		}
 	}
 
 	private async _submitCCU(ccuParams: Buffer): Promise<void> {
@@ -727,7 +735,6 @@ export class ChainConnectorPlugin extends BasePlugin<ChainConnectorPluginConfig>
 		 * TODO: As of now we save it in memory but going forward it should be saved in DB,
 		 * as the array size can grow after sometime.
 		 */
-		this._sentCCUs.push(tx);
 		// Save the sent CCU
 		const listOfCCUs = await this._chainConnectorStore.getListOfCCUs();
 		listOfCCUs.push(tx.toObject());

framework-plugins/lisk-framework-chain-connector-plugin/src/constants.ts

@@ -29,6 +29,7 @@ export const CCM_PROCESSED = 'ccmProcessed';
 export const CHAIN_ID_LENGTH = 4;
 export const DEFAULT_REGISTRATION_HEIGHT = 1;
 export const DEFAULT_LAST_CCM_SENT_NONCE = BigInt(-1);
+export const DEFAULT_CCU_SAVE_LIMIT = 300;
 
 export const DB_KEY_CROSS_CHAIN_MESSAGES = Buffer.from([1]);
 export const DB_KEY_BLOCK_HEADERS = Buffer.from([2]);

framework-plugins/lisk-framework-chain-connector-plugin/src/schemas.ts

@@ -13,7 +13,12 @@
  */
 
 import { chain, aggregateCommitSchema, ccmSchema } from 'lisk-sdk';
-import { CCU_FREQUENCY, CCU_TOTAL_CCM_SIZE, DEFAULT_REGISTRATION_HEIGHT } from './constants';
+import {
+	CCU_FREQUENCY,
+	CCU_TOTAL_CCM_SIZE,
+	DEFAULT_CCU_SAVE_LIMIT,
+	DEFAULT_REGISTRATION_HEIGHT,
+} from './constants';
 
 const pluginSchemaIDPrefix = '/lisk/plugins/chainConnector';
 
@@ -57,6 +62,11 @@ export const configSchema = {
 			minimum: 1,
 			maximum: CCU_TOTAL_CCM_SIZE,
 		},
+		ccuSaveLimit: {
+			type: 'integer',
+			description: 'Number of CCUs to save.',
+			minimum: 0,
+		},
 		registrationHeight: {
 			type: 'integer',
 			description: 'Height at the time of registration on the receiving chain.',
@@ -73,6 +83,7 @@ export const configSchema = {
 		isSaveCCU: false,
 		maxCCUSize: CCU_TOTAL_CCM_SIZE,
 		registrationHeight: DEFAULT_REGISTRATION_HEIGHT,
+		ccuSaveLimit: DEFAULT_CCU_SAVE_LIMIT,
 	},
 };
 

framework-plugins/lisk-framework-chain-connector-plugin/src/types.ts

@@ -36,6 +36,7 @@ export interface ChainConnectorPluginConfig {
 	isSaveCCU: boolean;
 	maxCCUSize: number;
 	registrationHeight: number;
+	ccuSaveLimit: number;
 }
 
 export type SentCCUs = Transaction[];

framework-plugins/lisk-framework-chain-connector-plugin/test/unit/plugin.spec.ts

@@ -55,6 +55,7 @@ import {
 import * as certificateGenerationUtil from '../../src/certificate_generation';
 import * as activeValidatorsUpdateUtil from '../../src/active_validators_update';
 import { getMainchainID } from '../../src/utils';
+import { getSampleCCU } from '../utils/sampleCCU';
 
 describe('ChainConnectorPlugin', () => {
 	const BLS_SIGNATURE_LENGTH = 96;
@@ -199,6 +200,7 @@ describe('ChainConnectorPlugin', () => {
 			ccuFrequency: CCU_FREQUENCY,
 			password: defaultPassword,
 			maxCCUSize: CCU_TOTAL_CCM_SIZE,
+			ccuSaveLimit: 1,
 			isSaveCCU: false,
 			registrationHeight: 1,
 			receivingChainID: getMainchainID(ownChainID).toString('hex'),
@@ -265,7 +267,7 @@ describe('ChainConnectorPlugin', () => {
 
 			await chainConnectorPlugin.load();
 
-			expect(chainConnectorPlugin['_receivingChainClient']).toBeDefined();
+			expect(chainConnectorPlugin['_receivingChainClient']).toBeUndefined();
 			expect(chainConnectorPlugin['_sendingChainClient']).toBe(chainConnectorPlugin['_apiClient']);
 		});
 
@@ -278,7 +280,7 @@ describe('ChainConnectorPlugin', () => {
 			chainConnectorPlugin['_apiClient'] = sendingChainAPIClientMock;
 			await chainConnectorPlugin.load();
 
-			expect(chainConnectorPlugin['_receivingChainClient']).toBeDefined();
+			expect(chainConnectorPlugin['_receivingChainClient']).toBeUndefined();
 			expect(chainConnectorPlugin['_sendingChainClient']).toBeDefined();
 		});
 
@@ -297,6 +299,14 @@ describe('ChainConnectorPlugin', () => {
 
 			chainConnectorPlugin['_apiClient'] = sendingChainAPIClientMock;
 			await chainConnectorPlugin.load();
+			jest.spyOn(chainConnectorPlugin as any, '_saveDataOnNewBlock').mockResolvedValue({});
+			await chainConnectorPlugin['_newBlockHandler']({
+				blockHeader: testing
+					.createFakeBlockHeader({
+						generatorAddress: Buffer.from('66687aadf862bd776c8fc18b8e9f8e2008971485', 'hex'),
+					})
+					.toJSON(),
+			});
 			expect(apiClient.createWSClient).toHaveBeenCalled();
 		});
 
@@ -312,9 +322,21 @@ describe('ChainConnectorPlugin', () => {
 				appConfig: appConfigForPlugin,
 			});
 			chainConnectorPlugin['_apiClient'] = sendingChainAPIClientMock;
+			await chainConnectorPlugin.load();
+			jest.spyOn(chainConnectorPlugin as any, '_saveDataOnNewBlock').mockResolvedValue({});
+			jest.spyOn(chainConnectorPlugin as any, '_initializeReceivingChainClient');
+			jest.spyOn(testing.mocks.loggerMock, 'error');
+			await chainConnectorPlugin['_newBlockHandler']({
+				blockHeader: testing
+					.createFakeBlockHeader({
+						generatorAddress: Buffer.from('66687aadf862bd776c8fc18b8e9f8e2008971485', 'hex'),
+					})
+					.toJSON(),
+			});
 
-			await expect(chainConnectorPlugin.load()).rejects.toThrow(
-				'IPC path and WS url are undefined in the configuration',
+			expect(testing.mocks.loggerMock.error).toHaveBeenCalledWith(
+				new Error('IPC path and WS url are undefined in the configuration.'),
+				'Failed while handling the new block',
 			);
 		});
 
@@ -341,11 +363,6 @@ describe('ChainConnectorPlugin', () => {
 
 			expect(sendingChainAPIClientMock.invoke).toHaveBeenCalledTimes(1);
 			expect(sendingChainAPIClientMock.subscribe).toHaveBeenCalledTimes(2);
-
-			expect(testing.mocks.loggerMock.error).toHaveBeenCalledWith(
-				new Error('IPC connection timed out.'),
-				'Not able to connect to receivingChainAPIClient. Trying again on next new block.',
-			);
 		});
 
 		it('should initialize _chainConnectorDB', async () => {
@@ -818,6 +835,7 @@ describe('ChainConnectorPlugin', () => {
 	describe('Cleanup Functions', () => {
 		let blockHeader1: BlockHeader;
 		let blockHeader2: BlockHeader;
+		let sampleCCUs: chain.TransactionAttrs[];
 
 		beforeEach(async () => {
 			await chainConnectorPlugin.init({
@@ -826,6 +844,12 @@ describe('ChainConnectorPlugin', () => {
 				appConfig: appConfigForPlugin,
 			});
 
+			await chainConnectorPlugin.init({
+				logger: testing.mocks.loggerMock,
+				config: defaultConfig,
+				appConfig: appConfigForPlugin,
+			});
+
 			when(sendingChainAPIClientMock.invoke)
 				.calledWith('interoperability_getOwnChainAccount')
 				.mockResolvedValue({
@@ -878,6 +902,8 @@ describe('ChainConnectorPlugin', () => {
 				blockHeader1,
 				blockHeader2,
 			] as never);
+			sampleCCUs = [getSampleCCU(), getSampleCCU()];
+			chainConnectorStoreMock.getListOfCCUs.mockResolvedValue(sampleCCUs as never);
 		});
 
 		it('should delete block headers with height less than _lastCertifiedHeight', async () => {
@@ -911,6 +937,16 @@ describe('ChainConnectorPlugin', () => {
 				chainConnectorPlugin['_chainConnectorStore'].setValidatorsHashPreimage,
 			).toHaveBeenCalledWith([]);
 		});
+
+		it('should delete sentCCUs based on config ccuSaveLimit', async () => {
+			await chainConnectorPlugin['_cleanup']();
+
+			expect(chainConnectorPlugin['_chainConnectorStore'].getListOfCCUs).toHaveBeenCalledTimes(1);
+
+			expect(chainConnectorPlugin['_chainConnectorStore'].setListOfCCUs).toHaveBeenCalledWith([
+				sampleCCUs[1],
+			]);
+		});
 	});
 
 	describe('_computeCCUParams', () => {
@@ -1002,6 +1038,7 @@ describe('ChainConnectorPlugin', () => {
 			chainConnectorPlugin['_apiClient'] = sendingChainAPIClientMock;
 
 			await chainConnectorPlugin.load();
+			chainConnectorPlugin['_receivingChainClient'] = receivingChainAPIClientMock;
 			// Set all the sample data
 			await chainConnectorPlugin['_chainConnectorStore'].setBlockHeaders(sampleBlockHeaders);
 			await chainConnectorPlugin['_chainConnectorStore'].setAggregateCommits(
@@ -1045,6 +1082,7 @@ describe('ChainConnectorPlugin', () => {
 					...getSampleCCM(12),
 					height: 10,
 				});
+
 				const result = await chainConnectorPlugin['_computeCCUParams'](
 					sampleBlockHeaders,
 					sampleAggregateCommits,

framework-plugins/lisk-framework-chain-connector-plugin/test/utils/sampleCCU.ts

@@ -0,0 +1,46 @@
+/*
+ * Copyright © 2023 Lisk Foundation
+ *
+ * See the LICENSE file at the top-level directory of this distribution
+ * for licensing information.
+ *
+ * Unless otherwise agreed in a custom licensing agreement with the Lisk Foundation,
+ * no part of this software, including this file, may be copied, modified,
+ * propagated, or distributed except according to the terms contained in the
+ * LICENSE file.
+ *
+ * Removal or modification of this copyright notice is prohibited.
+ */
+
+import {
+	testing,
+	SubmitMainchainCrossChainUpdateCommand,
+	MODULE_NAME_INTEROPERABILITY,
+} from 'lisk-sdk';
+
+export const getSampleCCU = (params?: Record<string, unknown>) =>
+	testing
+		.createTransaction({
+			commandClass: SubmitMainchainCrossChainUpdateCommand as any,
+			module: MODULE_NAME_INTEROPERABILITY,
+			params: params ?? {
+				activeValidatorsUpdate: {
+					blsKeysUpdate: [],
+					bftWeightsUpdate: [],
+					bftWeightsUpdateBitmap: Buffer.alloc(0),
+				},
+				certificate: Buffer.alloc(1),
+				certificateThreshold: BigInt(1),
+				inboxUpdate: {
+					crossChainMessages: [],
+					messageWitnessHashes: [],
+					outboxRootWitness: {
+						bitmap: Buffer.alloc(1),
+						siblingHashes: [],
+					},
+				},
+				sendingChainID: Buffer.from('04000001', 'hex'),
+			},
+			chainID: Buffer.from('04000001', 'hex'),
+		})
+		.toObject();

framework/src/modules/interoperability/base_interoperability_internal_methods.ts

@@ -401,7 +401,7 @@ export abstract class BaseInteroperabilityInternalMethod extends BaseInternalMet
 
 		const verifySignature = verifyAggregateCertificateSignature(
 			chainValidators.activeValidators,
-			params.certificateThreshold,
+			chainValidators.certificateThreshold,
 			params.sendingChainID,
 			certificate,
 		);

framework/test/unit/modules/interoperability/internal_method.spec.ts

@@ -1146,14 +1146,16 @@ describe('Base interoperability internal method', () => {
 			},
 		};
 
+		const chainValidators = {
+			activeValidators,
+			certificateThreshold: BigInt(20),
+		};
+
 		beforeEach(async () => {
 			jest.spyOn(interopMod.events.get(InvalidCertificateSignatureEvent), 'add');
 			await interopMod.stores
 				.get(ChainValidatorsStore)
-				.set(methodContext, txParams.sendingChainID, {
-					activeValidators,
-					certificateThreshold: BigInt(20),
-				});
+				.set(methodContext, txParams.sendingChainID, chainValidators);
 		});
 
 		it('should reject if verifyWeightedAggSig fails', async () => {
@@ -1171,7 +1173,7 @@ describe('Base interoperability internal method', () => {
 				txParams.sendingChainID,
 				encodedUnsignedCertificate,
 				activeValidators.map(v => v.bftWeight),
-				txParams.certificateThreshold,
+				chainValidators.certificateThreshold,
 			);
 
 			expect(interopMod.events.get(InvalidCertificateSignatureEvent).add).toHaveBeenCalledTimes(1);