Fix OOM error in large parentchain syncs

core/parentchain/block-import-dispatcher/src/triggered_dispatcher.rs

@@ -163,6 +163,7 @@ where
 		&self,
 		predicate: impl Fn(&BlockImporter::SignedBlockType) -> bool,
 	) -> Result<Option<BlockImporter::SignedBlockType>> {
+		trace!("Import of parentchain blocks and events has been triggered");
 		let blocks_to_import =
 			self.import_queue.pop_until(predicate).map_err(Error::ImportQueue)?;
 
@@ -174,7 +175,7 @@ where
 		let latest_imported_block = blocks_to_import.last().map(|b| (*b).clone());
 
 		trace!(
-			"Import of parentchain blocks and events has been triggered, importing {} blocks and {} events from queue",
+			"Importing {} blocks and {} events from queue",
 			blocks_to_import.len(),
 			events_to_import.len(),
 		);

service/src/account_funding.rs

@@ -146,7 +146,7 @@ fn bootstrap_funds_from_alice(
             "funding amount is too high: please change EXISTENTIAL_DEPOSIT_FACTOR_FOR_INIT_FUNDS ({:?})",
             funding_amount
         );
-		return Err(Error::ApplicationSetup)
+		return Err(Error::ApplicationSetup("Alice funds too low".into()))
 	}
 
 	let mut alice_signer_api = api.clone();

service/src/error.rs

@@ -38,8 +38,8 @@ pub enum Error {
 	Serialization(#[from] serde_json::Error),
 	#[error("{0}")]
 	FromUtf8(#[from] std::string::FromUtf8Error),
-	#[error("Application setup error!")]
-	ApplicationSetup,
+	#[error("Application setup error: {0}")]
+	ApplicationSetup(String),
 	#[error("Failed to find any peer worker")]
 	NoPeerWorkerFound,
 	#[error("No worker for shard {0} found on parentchain")]
@@ -50,6 +50,8 @@ pub enum Error {
 	MissingGenesisHeader,
 	#[error("Could not find last finalized block of the parentchain")]
 	MissingLastFinalizedBlock,
+	#[error("Error during the block sync process: {0}")]
+	BlockSync(String),
 	#[error("{0}")]
 	Custom(Box<dyn std::error::Error + Sync + Send + 'static>),
 }

service/src/main.rs

@@ -387,9 +387,6 @@ fn start_worker<E, T, D, InitializationHandler, WorkerModeProvider>(
 
 	let tokio_handle = tokio_handle_getter.get_handle();
 
-	#[cfg(feature = "teeracle")]
-	let teeracle_tokio_handle = tokio_handle.clone();
-
 	// ------------------------------------------------------------------------
 	// Get the public key of our TEE.
 	let tee_accountid = enclave_account(enclave.as_ref());
@@ -455,7 +452,7 @@ fn start_worker<E, T, D, InitializationHandler, WorkerModeProvider>(
 			&config,
 			enclave.clone(),
 			sidechain_storage.clone(),
-			tokio_handle,
+			&tokio_handle,
 		);
 	}
 
@@ -542,34 +539,39 @@ fn start_worker<E, T, D, InitializationHandler, WorkerModeProvider>(
 
 	initialization_handler.registered_on_parentchain();
 
-	// ------------------------------------------------------------------------
-	// initialize teeracle interval
-	#[cfg(feature = "teeracle")]
-	if WorkerModeProvider::worker_mode() == WorkerMode::Teeracle {
-		schedule_periodic_reregistration_thread(
-			send_register_xt,
-			run_config.reregister_teeracle_interval(),
-		);
+	match WorkerModeProvider::worker_mode() {
+		WorkerMode::Teeracle => {
+			// ------------------------------------------------------------------------
+			// initialize teeracle interval
+			#[cfg(feature = "teeracle")]
+			schedule_periodic_reregistration_thread(
+				send_register_xt,
+				run_config.reregister_teeracle_interval(),
+			);
 
-		start_periodic_market_update(
-			&integritee_rpc_api,
-			run_config.teeracle_update_interval(),
-			enclave.as_ref(),
-			&teeracle_tokio_handle,
-		);
-	}
+			#[cfg(feature = "teeracle")]
+			start_periodic_market_update(
+				&integritee_rpc_api,
+				run_config.teeracle_update_interval(),
+				enclave.as_ref(),
+				&tokio_handle.clone(),
+			);
+		},
+		WorkerMode::OffChainWorker => {
+			println!("*** [+] Finished initializing light client, syncing parentchain...");
 
-	if WorkerModeProvider::worker_mode() != WorkerMode::Teeracle {
-		println!("*** [+] Finished initializing light client, syncing parentchain...");
+			// Syncing all parentchain blocks, this might take a while..
+			let last_synced_header =
+				parentchain_handler.sync_parentchain(last_synced_header).unwrap();
 
-		// Syncing all parentchain blocks, this might take a while..
-		let mut last_synced_header =
-			parentchain_handler.sync_parentchain(last_synced_header).unwrap();
+			start_parentchain_header_subscription_thread(parentchain_handler, last_synced_header);
+		},
+		WorkerMode::Sidechain => {
+			println!("*** [+] Finished initializing light client, syncing parentchain...");
 
-		// ------------------------------------------------------------------------
-		// Initialize the sidechain
-		if WorkerModeProvider::worker_mode() == WorkerMode::Sidechain {
-			last_synced_header = sidechain_init_block_production(
+			// ------------------------------------------------------------------------
+			// Initialize the sidechain
+			let last_synced_header = sidechain_init_block_production(
 				enclave.clone(),
 				&register_enclave_xt_header,
 				we_are_primary_validateer,
@@ -578,26 +580,15 @@ fn start_worker<E, T, D, InitializationHandler, WorkerModeProvider>(
 				&last_synced_header,
 			)
 			.unwrap();
-		}
 
-		// ------------------------------------------------------------------------
-		// start parentchain syncing loop (subscribe to header updates)
-		thread::Builder::new()
-			.name("parentchain_sync_loop".to_owned())
-			.spawn(move || {
-				if let Err(e) =
-					subscribe_to_parentchain_new_headers(parentchain_handler, last_synced_header)
-				{
-					error!("Parentchain block syncing terminated with a failure: {:?}", e);
-				}
-				println!("[!] Parentchain block syncing has terminated");
-			})
-			.unwrap();
-	}
+			start_parentchain_header_subscription_thread(parentchain_handler, last_synced_header);
 
-	// ------------------------------------------------------------------------
-	if WorkerModeProvider::worker_mode() == WorkerMode::Sidechain {
-		spawn_worker_for_shard_polling(shard, integritee_rpc_api.clone(), initialization_handler);
+			spawn_worker_for_shard_polling(
+				shard,
+				integritee_rpc_api.clone(),
+				initialization_handler,
+			);
+		},
 	}
 
 	if let Some(url) = config.target_a_parentchain_rpc_endpoint() {
@@ -666,20 +657,7 @@ fn init_target_parentchain<E>(
 			parentchain_handler.sync_parentchain(last_synched_header).unwrap();
 
 		// start parentchain syncing loop (subscribe to header updates)
-		thread::Builder::new()
-			.name(format!("{:?}_parentchain_sync_loop", parentchain_id))
-			.spawn(move || {
-				if let Err(e) =
-					subscribe_to_parentchain_new_headers(parentchain_handler, last_synched_header)
-				{
-					error!(
-						"[{:?}] parentchain block syncing terminated with a failure: {:?}",
-						parentchain_id, e
-					);
-				}
-				println!("[!] [{:?}] parentchain block syncing has terminated", parentchain_id);
-			})
-			.unwrap();
+		start_parentchain_header_subscription_thread(parentchain_handler, last_synched_header)
 	}
 
 	// Subscribe to events and print them.
@@ -1042,6 +1020,27 @@ fn send_extrinsic(
 	}
 }
 
+fn start_parentchain_header_subscription_thread<E: EnclaveBase + Sidechain>(
+	parentchain_handler: Arc<ParentchainHandler<ParentchainApi, E>>,
+	last_synced_header: Header,
+) {
+	let parentchain_id = parentchain_handler.parentchain_id().clone();
+	thread::Builder::new()
+		.name(format!("{:?}_parentchain_sync_loop", parentchain_id))
+		.spawn(move || {
+			if let Err(e) =
+				subscribe_to_parentchain_new_headers(parentchain_handler, last_synced_header)
+			{
+				error!(
+					"[{:?}] parentchain block syncing terminated with a failure: {:?}",
+					parentchain_id, e
+				);
+			}
+			println!("[!] [{:?}] parentchain block syncing has terminated", parentchain_id);
+		})
+		.unwrap();
+}
+
 /// Subscribe to the node API finalized heads stream and trigger a parent chain sync
 /// upon receiving a new header.
 fn subscribe_to_parentchain_new_headers<E: EnclaveBase + Sidechain>(