fix restart issue of staking miner (#4073)

Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>

polkadot/utils/staking-miner/src/dry_run.rs

@@ -117,11 +117,11 @@ macro_rules! dry_run_cmd_for { ($runtime:ident) => { paste::paste! {
 		let mut ext = crate::create_election_ext::<Runtime, Block>(
 			shared.uri.clone(),
 			config.at,
-			vec!["Staking".to_string(), "System".to_string(), "Balances".to_string()]
+			vec!["Staking".to_string(), "System".to_string()],
 		).await?;
 		force_create_snapshot::<Runtime>(&mut ext)?;
 
-		let (raw_solution, witness) = crate::mine_with::<Runtime>(&config.solver, &mut ext)?;
+		let (raw_solution, witness) = crate::mine_with::<Runtime>(&config.solver, &mut ext, false)?;
 
 		let nonce = crate::get_account_info::<Runtime>(client, &signer.account, config.at)
 			.await?

polkadot/utils/staking-miner/src/emergency_solution.rs

@@ -16,7 +16,7 @@
 
 //! The emergency-solution command.
 
-use crate::{prelude::*, Error, SharedConfig};
+use crate::{prelude::*, EmergencySolutionConfig, Error, SharedConfig};
 use codec::Encode;
 use frame_election_provider_support::SequentialPhragmen;
 use std::io::Write;
@@ -25,25 +25,35 @@ macro_rules! emergency_solution_cmd_for { ($runtime:ident) => { paste::paste! {
 	/// Execute the emergency-solution command.
 	pub(crate) async fn [<emergency_solution_cmd_ $runtime>](
 		shared: SharedConfig,
+		config: EmergencySolutionConfig,
 	) -> Result<(), Error<$crate::[<$runtime _runtime_exports>]::Runtime>> {
 		use $crate::[<$runtime _runtime_exports>]::*;
 		let mut ext = crate::create_election_ext::<Runtime, Block>(shared.uri.clone(), None, vec![]).await?;
 		ext.execute_with(|| {
 			assert!(EPM::Pallet::<Runtime>::current_phase().is_emergency());
+
 			// NOTE: this internally calls feasibility_check, but we just re-do it here as an easy way
 			// to get a `ReadySolution`.
 			let (raw_solution, _) =
 				<EPM::Pallet<Runtime>>::mine_solution::<SequentialPhragmen<AccountId, sp_runtime::Perbill>>()?;
 			log::info!(target: LOG_TARGET, "mined solution with {:?}", &raw_solution.score);
-			let ready_solution = EPM::Pallet::<Runtime>::feasibility_check(raw_solution, EPM::ElectionCompute::Signed)?;
+			let mut ready_solution = EPM::Pallet::<Runtime>::feasibility_check(raw_solution, EPM::ElectionCompute::Signed)?;
-			let encoded_ready = ready_solution.encode();
+
+			// maybe truncate.
+			if let Some(take) = config.take {
+				log::info!(target: LOG_TARGET, "truncating {} winners to {}", ready_solution.supports.len(), take);
+				ready_solution.supports.sort_unstable_by_key(|(_, s)| s.total);
+				ready_solution.supports.truncate(take);
+			}
+
+			// write to file and stdout.
 			let encoded_support = ready_solution.supports.encode();
-			let mut solution_file = std::fs::File::create("solution.bin")?;
 			let mut supports_file = std::fs::File::create("solution.supports.bin")?;
-			solution_file.write_all(&encoded_ready)?;
 			supports_file.write_all(&encoded_support)?;
-			log::info!(target: LOG_TARGET, "ReadySolution: size {:?} / score = {:?}", encoded_ready.len(), ready_solution.score);
+
+			log::info!(target: LOG_TARGET, "ReadySolution: size {:?} / score = {:?}", ready_solution.encoded_size(), ready_solution.score);
 			log::trace!(target: LOG_TARGET, "Supports: {}", sp_core::hexdisplay::HexDisplay::from(&encoded_support));
+
 			Ok(())
 		})
 	}

polkadot/utils/staking-miner/src/main.rs

@@ -276,7 +276,7 @@ enum Command {
 	/// Just compute a solution now, and don't submit it.
 	DryRun(DryRunConfig),
 	/// Provide a solution that can be submitted to the chain as an emergency response.
-	EmergencySolution,
+	EmergencySolution(EmergencySolutionConfig),
 }
 
 #[derive(Debug, Clone, StructOpt)]
@@ -291,39 +291,6 @@ enum Solvers {
 	},
 }
 
-/// Mine a solution with the given `solver`.
-fn mine_with<T>(
-	solver: &Solvers,
-	ext: &mut Ext,
-) -> Result<(EPM::RawSolution<EPM::SolutionOf<T>>, u32), Error<T>>
-where
-	T: EPM::Config,
-	T::Solver: NposSolver<Error = sp_npos_elections::Error>,
-{
-	use frame_election_provider_support::{PhragMMS, SequentialPhragmen};
-
-	match solver {
-		Solvers::SeqPhragmen { iterations } => {
-			BalanceIterations::set(*iterations);
-			mine_unchecked::<
-				T,
-				SequentialPhragmen<
-					<T as frame_system::Config>::AccountId,
-					sp_runtime::Perbill,
-					Balancing,
-				>,
-			>(ext, false)
-		},
-		Solvers::PhragMMS { iterations } => {
-			BalanceIterations::set(*iterations);
-			mine_unchecked::<
-				T,
-				PhragMMS<<T as frame_system::Config>::AccountId, sp_runtime::Perbill, Balancing>,
-			>(ext, false)
-		},
-	}
-}
-
 frame_support::parameter_types! {
 	/// Number of balancing iterations for a solution algorithm. Set based on the [`Solvers`] CLI
 	/// config.
@@ -341,16 +308,32 @@ struct MonitorConfig {
 	#[structopt(long, default_value = "head", possible_values = &["head", "finalized"])]
 	listen: String,
 
+	/// The solver algorithm to use.
 	#[structopt(subcommand)]
 	solver: Solvers,
 }
 
+#[derive(Debug, Clone, StructOpt)]
+struct EmergencySolutionConfig {
+	/// The block hash at which scraping happens. If none is provided, the latest head is used.
+	#[structopt(long)]
+	at: Option<Hash>,
+
+	/// The solver algorithm to use.
+	#[structopt(subcommand)]
+	solver: Solvers,
+
+	/// The number of top backed winners to take. All are taken, if not provided.
+	take: Option<usize>,
+}
+
 #[derive(Debug, Clone, StructOpt)]
 struct DryRunConfig {
 	/// The block hash at which scraping happens. If none is provided, the latest head is used.
 	#[structopt(long)]
 	at: Option<Hash>,
 
+	/// The solver algorithm to use.
 	#[structopt(subcommand)]
 	solver: Solvers,
 }
@@ -407,9 +390,9 @@ async fn create_election_ext<T: EPM::Config, B: BlockT>(
 		.map_err(|why| Error::RemoteExternalities(why))
 }
 
-/// Compute the election at the given block number. It expects to NOT be `Phase::Off`. In other
+/// Compute the election. It expects to NOT be `Phase::Off`. In other words, the snapshot must
-/// words, the snapshot must exists on the given externalities.
+/// exists on the given externalities.
-fn mine_unchecked<T, S>(
+fn mine_solution<T, S>(
 	ext: &mut Ext,
 	do_feasibility: bool,
 ) -> Result<(EPM::RawSolution<EPM::SolutionOf<T>>, u32), Error<T>>
@@ -434,6 +417,40 @@ where
 	})
 }
 
+/// Mine a solution with the given `solver`.
+fn mine_with<T>(
+	solver: &Solvers,
+	ext: &mut Ext,
+	do_feasibility: bool,
+) -> Result<(EPM::RawSolution<EPM::SolutionOf<T>>, u32), Error<T>>
+where
+	T: EPM::Config,
+	T::Solver: NposSolver<Error = sp_npos_elections::Error>,
+{
+	use frame_election_provider_support::{PhragMMS, SequentialPhragmen};
+
+	match solver {
+		Solvers::SeqPhragmen { iterations } => {
+			BalanceIterations::set(*iterations);
+			mine_solution::<
+				T,
+				SequentialPhragmen<
+					<T as frame_system::Config>::AccountId,
+					sp_runtime::Perbill,
+					Balancing,
+				>,
+			>(ext, do_feasibility)
+		},
+		Solvers::PhragMMS { iterations } => {
+			BalanceIterations::set(*iterations);
+			mine_solution::<
+				T,
+				PhragMMS<<T as frame_system::Config>::AccountId, sp_runtime::Perbill, Balancing>,
+			>(ext, do_feasibility)
+		},
+	}
+}
+
 #[allow(unused)]
 fn mine_dpos<T: EPM::Config>(ext: &mut Ext) -> Result<(), Error<T>> {
 	ext.execute_with(|| {
@@ -474,7 +491,6 @@ fn mine_dpos<T: EPM::Config>(ext: &mut Ext) -> Result<(), Error<T>> {
 
 pub(crate) async fn check_versions<T: frame_system::Config + EPM::Config>(
 	client: &WsClient,
-	print: bool,
 ) -> Result<(), Error<T>> {
 	let linked_version = T::Version::get();
 	let on_chain_version = rpc_helpers::rpc::<sp_version::RuntimeVersion>(
@@ -485,10 +501,9 @@ pub(crate) async fn check_versions<T: frame_system::Config + EPM::Config>(
 	.await
 	.expect("runtime version RPC should always work; qed");
 
-	if print {
+	log::debug!(target: LOG_TARGET, "linked version {:?}", linked_version);
-		log::info!(target: LOG_TARGET, "linked version {:?}", linked_version);
+	log::debug!(target: LOG_TARGET, "on-chain version {:?}", on_chain_version);
-		log::info!(target: LOG_TARGET, "on-chain version {:?}", on_chain_version);
+
-	}
 	if linked_version != on_chain_version {
 		log::error!(
 			target: LOG_TARGET,
@@ -576,7 +591,7 @@ async fn main() {
 	log::info!(target: LOG_TARGET, "connected to chain {:?}", chain);
 
 	any_runtime_unit! {
-		check_versions::<Runtime>(&client, true).await
+		check_versions::<Runtime>(&client).await
 	};
 
 	let signer_account = any_runtime! {
@@ -595,7 +610,7 @@ async fn main() {
 				.map_err(|e| {
 					log::error!(target: LOG_TARGET, "DryRun error: {:?}", e);
 				}),
-			Command::EmergencySolution => emergency_solution_cmd(shared.clone()).await
+			Command::EmergencySolution(c) => emergency_solution_cmd(shared.clone(), c).await
 				.map_err(|e| {
 					log::error!(target: LOG_TARGET, "EmergencySolution error: {:?}", e);
 				}),

polkadot/utils/staking-miner/src/monitor.rs

@@ -21,7 +21,7 @@ use crate::{
 };
 use codec::Encode;
 use jsonrpsee_ws_client::{
-	types::{traits::SubscriptionClient, v2::params::JsonRpcParams, Subscription},
+	types::{traits::SubscriptionClient, Subscription},
 	WsClient,
 };
 use sc_transaction_pool_api::TransactionStatus;
@@ -71,106 +71,113 @@ macro_rules! monitor_cmd_for { ($runtime:tt) => { paste::paste! {
 		signer: Signer,
 	) -> Result<(), Error<$crate::[<$runtime _runtime_exports>]::Runtime>> {
 		use $crate::[<$runtime _runtime_exports>]::*;
+
 		let (sub, unsub) = if config.listen == "head" {
 			("chain_subscribeNewHeads", "chain_unsubscribeNewHeads")
 		} else {
 			("chain_subscribeFinalizedHeads", "chain_unsubscribeFinalizedHeads")
 		};
 
-		log::info!(target: LOG_TARGET, "subscribing to {:?} / {:?}", sub, unsub);
+		loop {
-		let mut subscription: Subscription<Header> = client
+			log::info!(target: LOG_TARGET, "subscribing to {:?} / {:?}", sub, unsub);
-			.subscribe(&sub, JsonRpcParams::NoParams, &unsub)
+			let mut subscription: Subscription<Header> = client
-			.await
+				.subscribe(&sub, params! {}, &unsub)
-			.unwrap();
+				.await
+				.unwrap();
 
-		while let Some(now) = subscription.next().await? {
+			while let Some(now) = subscription.next().await? {
-			let hash = now.hash();
+				let hash = now.hash();
-			log::debug!(target: LOG_TARGET, "new event at #{:?} ({:?})", now.number, hash);
+				log::trace!(target: LOG_TARGET, "new event at #{:?} ({:?})", now.number, hash);
 
-			// if the runtime version has changed, terminate
+				// if the runtime version has changed, terminate.
-			crate::check_versions::<Runtime>(client, false).await?;
+				crate::check_versions::<Runtime>(client).await?;
 
-			// we prefer doing this check before fetching anything into a remote-ext.
+				// we prefer doing this check before fetching anything into a remote-ext.
-			if ensure_signed_phase::<Runtime, Block>(client, hash).await.is_err() {
+				if ensure_signed_phase::<Runtime, Block>(client, hash).await.is_err() {
-				log::debug!(target: LOG_TARGET, "phase closed, not interested in this block at all.");
+					log::debug!(target: LOG_TARGET, "phase closed, not interested in this block at all.");
-				continue;
+					continue;
-			};
+				};
 
-			// NOTE: we don't check the score of any of the submitted solutions. If we submit a weak
+				// grab an externalities without staking, just the election snapshot.
-			// one, as long as we are valid, we will end up getting our deposit back, so not a big
+				let mut ext = crate::create_election_ext::<Runtime, Block>(
-			// deal for now. Note that to avoid an unfeasible solution, we should make sure that we
+					shared.uri.clone(),
-			// only start the process on a finalized snapshot. If the signed phase is long enough,
+					Some(hash),
-			// this will not be a solution.
+					vec![],
+				).await?;
 
-			// grab an externalities without staking, just the election snapshot.
+				if ensure_no_previous_solution::<Runtime, Block>(&mut ext, &signer.account).await.is_err() {
-			let mut ext = crate::create_election_ext::<Runtime, Block>(shared.uri.clone(), Some(hash), vec![]).await?;
+					log::debug!(target: LOG_TARGET, "We already have a solution in this phase, skipping.");
+					continue;
+				}
 
-			if ensure_no_previous_solution::<Runtime, Block>(&mut ext, &signer.account).await.is_err() {
+				// mine a solution, and run feasibility check on it as well.
-				log::debug!(target: LOG_TARGET, "We already have a solution in this phase, skipping.");
+				let (raw_solution, witness) = crate::mine_with::<Runtime>(&config.solver, &mut ext, true)?;
-				continue;
+				log::info!(target: LOG_TARGET, "mined solution with {:?}", &raw_solution.score);
+
+				let nonce = crate::get_account_info::<Runtime>(client, &signer.account, Some(hash))
+					.await?
+					.map(|i| i.nonce)
+					.expect(crate::signer::SIGNER_ACCOUNT_WILL_EXIST);
+				let tip = 0 as Balance;
+				let period = <Runtime as frame_system::Config>::BlockHashCount::get() / 2;
+				let current_block = now.number.saturating_sub(1);
+				let era = sp_runtime::generic::Era::mortal(period.into(), current_block.into());
+				log::trace!(
+					target: LOG_TARGET, "transaction mortality: {:?} -> {:?}",
+					era.birth(current_block.into()),
+					era.death(current_block.into()),
+				);
+				let extrinsic = ext.execute_with(|| create_uxt(raw_solution, witness, signer.clone(), nonce, tip, era));
+				let bytes = sp_core::Bytes(extrinsic.encode());
+
+				let mut tx_subscription: Subscription<
+					TransactionStatus<<Block as BlockT>::Hash, <Block as BlockT>::Hash>
+				> = match client
+					.subscribe(&"author_submitAndWatchExtrinsic", params! { bytes }, "author_unwatchExtrinsic")
+					.await
+				{
+					Ok(sub) => sub,
+					Err(why) => {
+					// This usually happens when we've been busy with mining for a few blocks, and
+					// now we're receiving the subscriptions of blocks in which we were busy. In
+					// these blocks, we still don't have a solution, so we re-compute a new solution
+					// and submit it with an outdated `Nonce`, which yields most often `Stale`
+					// error. NOTE: to improve this overall, and to be able to introduce an array of
+					// other fancy features, we should make this multi-threaded and do the
+					// computation outside of this callback.
+						log::warn!(target: LOG_TARGET, "failing to submit a transaction {:?}. continuing...", why);
+						continue
+					}
+				};
+
+				while let Some(status_update) = tx_subscription.next().await? {
+					log::trace!(target: LOG_TARGET, "status update {:?}", status_update);
+					match status_update {
+						TransactionStatus::Ready | TransactionStatus::Broadcast(_) | TransactionStatus::Future => continue,
+						TransactionStatus::InBlock(hash) => {
+							log::info!(target: LOG_TARGET, "included at {:?}", hash);
+							let key = frame_support::storage::storage_prefix(b"System", b"Events");
+							let events = get_storage::<Vec<frame_system::EventRecord<Event, <Block as BlockT>::Hash>>,
+							>(client, params!{ key, hash }).await?.unwrap_or_default();
+							log::info!(target: LOG_TARGET, "events at inclusion {:?}", events);
+						}
+						TransactionStatus::Retracted(hash) => {
+							log::info!(target: LOG_TARGET, "Retracted at {:?}", hash);
+						}
+						TransactionStatus::Finalized(hash) => {
+							log::info!(target: LOG_TARGET, "Finalized at {:?}", hash);
+							break
+						}
+						_ => {
+							log::warn!(target: LOG_TARGET, "Stopping listen due to other status {:?}", status_update);
+							break
+						}
+					}
+				};
 			}
 
-			let (raw_solution, witness) = crate::mine_with::<Runtime>(&config.solver, &mut ext)?;
+			log::warn!(target: LOG_TARGET, "subscription to {} terminated. Retrying..", sub)
-
-			log::info!(target: LOG_TARGET, "mined solution with {:?}", &raw_solution.score);
-
-			let nonce = crate::get_account_info::<Runtime>(client, &signer.account, Some(hash))
-				.await?
-				.map(|i| i.nonce)
-				.expect(crate::signer::SIGNER_ACCOUNT_WILL_EXIST);
-			let tip = 0 as Balance;
-			let period = <Runtime as frame_system::Config>::BlockHashCount::get() / 2;
-			let current_block = now.number.saturating_sub(1);
-			let era = sp_runtime::generic::Era::mortal(period.into(), current_block.into());
-			log::trace!(target: LOG_TARGET, "transaction mortality: {:?} -> {:?}", era.birth(current_block.into()), era.death(current_block.into()));
-			let extrinsic = ext.execute_with(|| create_uxt(raw_solution, witness, signer.clone(), nonce, tip, era));
-			let bytes = sp_core::Bytes(extrinsic.encode());
-
-			let mut tx_subscription: Subscription<
-				TransactionStatus<<Block as BlockT>::Hash, <Block as BlockT>::Hash>
-			> = match client
-				.subscribe(&"author_submitAndWatchExtrinsic", params! { bytes }, "author_unwatchExtrinsic")
-				.await
-			{
-				Ok(sub) => sub,
-				Err(why) => {
-				// This usually happens when we've been busy with mining for a few blocks, and now we're receiving the
-				// subscriptions of blocks in which we were busy. In these blocks, we still don't have a solution, so we
-				// re-compute a new solution and submit it with an outdated `Nonce`, which yields most often `Stale`
-				// error. NOTE: to improve this overall, and to be able to introduce an array of other fancy features,
-				// we should make this multi-threaded and do the computation outside of this callback.
-					log::warn!(target: LOG_TARGET, "failing to submit a transaction {:?}. continuing...", why);
-					continue
-				}
-			};
-
-			let _success = while let Some(status_update) = tx_subscription.next().await? {
-				log::trace!(target: LOG_TARGET, "status update {:?}", status_update);
-				match status_update {
-					TransactionStatus::Ready | TransactionStatus::Broadcast(_) | TransactionStatus::Future => continue,
-					TransactionStatus::InBlock(hash) => {
-						log::info!(target: LOG_TARGET, "included at {:?}", hash);
-						let key = frame_support::storage::storage_prefix(b"System", b"Events");
-						let events = get_storage::<Vec<frame_system::EventRecord<Event, <Block as BlockT>::Hash>>,
-						>(client, params!{ key, hash }).await?.unwrap_or_default();
-						log::info!(target: LOG_TARGET, "events at inclusion {:?}", events);
-					}
-					TransactionStatus::Retracted(hash) => {
-						log::info!(target: LOG_TARGET, "Retracted at {:?}", hash);
-					}
-					TransactionStatus::Finalized(hash) => {
-						log::info!(target: LOG_TARGET, "Finalized at {:?}", hash);
-						break
-					}
-					_ => {
-						log::warn!(target: LOG_TARGET, "Stopping listen due to other status {:?}", status_update);
-						break
-					}
-				}
-			};
 		}
-
-		Ok(())
 	}
 }}}