Unverified Commit 18445197 authored by Kian Paimani's avatar Kian Paimani Committed by GitHub
Browse files

fix restart issue of staking miner (#4073)


Co-authored-by: Niklas Adolfsson's avatarNiklas Adolfsson <niklasadolfsson1@gmail.com>
parent 90640249
Pipeline #162109 canceled with stages
in 7 minutes and 58 seconds
...@@ -117,11 +117,11 @@ macro_rules! dry_run_cmd_for { ($runtime:ident) => { paste::paste! { ...@@ -117,11 +117,11 @@ macro_rules! dry_run_cmd_for { ($runtime:ident) => { paste::paste! {
let mut ext = crate::create_election_ext::<Runtime, Block>( let mut ext = crate::create_election_ext::<Runtime, Block>(
shared.uri.clone(), shared.uri.clone(),
config.at, config.at,
vec!["Staking".to_string(), "System".to_string(), "Balances".to_string()] vec!["Staking".to_string(), "System".to_string()],
).await?; ).await?;
force_create_snapshot::<Runtime>(&mut ext)?; 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) let nonce = crate::get_account_info::<Runtime>(client, &signer.account, config.at)
.await? .await?
......
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
//! The emergency-solution command. //! The emergency-solution command.
use crate::{prelude::*, Error, SharedConfig}; use crate::{prelude::*, EmergencySolutionConfig, Error, SharedConfig};
use codec::Encode; use codec::Encode;
use frame_election_provider_support::SequentialPhragmen; use frame_election_provider_support::SequentialPhragmen;
use std::io::Write; use std::io::Write;
...@@ -25,25 +25,35 @@ macro_rules! emergency_solution_cmd_for { ($runtime:ident) => { paste::paste! { ...@@ -25,25 +25,35 @@ macro_rules! emergency_solution_cmd_for { ($runtime:ident) => { paste::paste! {
/// Execute the emergency-solution command. /// Execute the emergency-solution command.
pub(crate) async fn [<emergency_solution_cmd_ $runtime>]( pub(crate) async fn [<emergency_solution_cmd_ $runtime>](
shared: SharedConfig, shared: SharedConfig,
config: EmergencySolutionConfig,
) -> Result<(), Error<$crate::[<$runtime _runtime_exports>]::Runtime>> { ) -> Result<(), Error<$crate::[<$runtime _runtime_exports>]::Runtime>> {
use $crate::[<$runtime _runtime_exports>]::*; use $crate::[<$runtime _runtime_exports>]::*;
let mut ext = crate::create_election_ext::<Runtime, Block>(shared.uri.clone(), None, vec![]).await?; let mut ext = crate::create_election_ext::<Runtime, Block>(shared.uri.clone(), None, vec![]).await?;
ext.execute_with(|| { ext.execute_with(|| {
assert!(EPM::Pallet::<Runtime>::current_phase().is_emergency()); 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 // NOTE: this internally calls feasibility_check, but we just re-do it here as an easy way
// to get a `ReadySolution`. // to get a `ReadySolution`.
let (raw_solution, _) = let (raw_solution, _) =
<EPM::Pallet<Runtime>>::mine_solution::<SequentialPhragmen<AccountId, sp_runtime::Perbill>>()?; <EPM::Pallet<Runtime>>::mine_solution::<SequentialPhragmen<AccountId, sp_runtime::Perbill>>()?;
log::info!(target: LOG_TARGET, "mined solution with {:?}", &raw_solution.score); 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 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")?; let mut supports_file = std::fs::File::create("solution.supports.bin")?;
solution_file.write_all(&encoded_ready)?;
supports_file.write_all(&encoded_support)?; 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)); log::trace!(target: LOG_TARGET, "Supports: {}", sp_core::hexdisplay::HexDisplay::from(&encoded_support));
Ok(()) Ok(())
}) })
} }
......
...@@ -276,7 +276,7 @@ enum Command { ...@@ -276,7 +276,7 @@ enum Command {
/// Just compute a solution now, and don't submit it. /// Just compute a solution now, and don't submit it.
DryRun(DryRunConfig), DryRun(DryRunConfig),
/// Provide a solution that can be submitted to the chain as an emergency response. /// Provide a solution that can be submitted to the chain as an emergency response.
EmergencySolution, EmergencySolution(EmergencySolutionConfig),
} }
#[derive(Debug, Clone, StructOpt)] #[derive(Debug, Clone, StructOpt)]
...@@ -291,39 +291,6 @@ enum Solvers { ...@@ -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! { frame_support::parameter_types! {
/// Number of balancing iterations for a solution algorithm. Set based on the [`Solvers`] CLI /// Number of balancing iterations for a solution algorithm. Set based on the [`Solvers`] CLI
/// config. /// config.
...@@ -341,16 +308,32 @@ struct MonitorConfig { ...@@ -341,16 +308,32 @@ struct MonitorConfig {
#[structopt(long, default_value = "head", possible_values = &["head", "finalized"])] #[structopt(long, default_value = "head", possible_values = &["head", "finalized"])]
listen: String, listen: String,
/// The solver algorithm to use.
#[structopt(subcommand)] #[structopt(subcommand)]
solver: Solvers, 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)] #[derive(Debug, Clone, StructOpt)]
struct DryRunConfig { struct DryRunConfig {
/// The block hash at which scraping happens. If none is provided, the latest head is used. /// The block hash at which scraping happens. If none is provided, the latest head is used.
#[structopt(long)] #[structopt(long)]
at: Option<Hash>, at: Option<Hash>,
/// The solver algorithm to use.
#[structopt(subcommand)] #[structopt(subcommand)]
solver: Solvers, solver: Solvers,
} }
...@@ -407,9 +390,9 @@ async fn create_election_ext<T: EPM::Config, B: BlockT>( ...@@ -407,9 +390,9 @@ async fn create_election_ext<T: EPM::Config, B: BlockT>(
.map_err(|why| Error::RemoteExternalities(why)) .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, ext: &mut Ext,
do_feasibility: bool, do_feasibility: bool,
) -> Result<(EPM::RawSolution<EPM::SolutionOf<T>>, u32), Error<T>> ) -> Result<(EPM::RawSolution<EPM::SolutionOf<T>>, u32), Error<T>>
...@@ -434,6 +417,40 @@ where ...@@ -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)] #[allow(unused)]
fn mine_dpos<T: EPM::Config>(ext: &mut Ext) -> Result<(), Error<T>> { fn mine_dpos<T: EPM::Config>(ext: &mut Ext) -> Result<(), Error<T>> {
ext.execute_with(|| { ext.execute_with(|| {
...@@ -474,7 +491,6 @@ fn mine_dpos<T: EPM::Config>(ext: &mut Ext) -> Result<(), Error<T>> { ...@@ -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>( pub(crate) async fn check_versions<T: frame_system::Config + EPM::Config>(
client: &WsClient, client: &WsClient,
print: bool,
) -> Result<(), Error<T>> { ) -> Result<(), Error<T>> {
let linked_version = T::Version::get(); let linked_version = T::Version::get();
let on_chain_version = rpc_helpers::rpc::<sp_version::RuntimeVersion>( 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>( ...@@ -485,10 +501,9 @@ pub(crate) async fn check_versions<T: frame_system::Config + EPM::Config>(
.await .await
.expect("runtime version RPC should always work; qed"); .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 { if linked_version != on_chain_version {
log::error!( log::error!(
target: LOG_TARGET, target: LOG_TARGET,
...@@ -576,7 +591,7 @@ async fn main() { ...@@ -576,7 +591,7 @@ async fn main() {
log::info!(target: LOG_TARGET, "connected to chain {:?}", chain); log::info!(target: LOG_TARGET, "connected to chain {:?}", chain);
any_runtime_unit! { any_runtime_unit! {
check_versions::<Runtime>(&client, true).await check_versions::<Runtime>(&client).await
}; };
let signer_account = any_runtime! { let signer_account = any_runtime! {
...@@ -595,7 +610,7 @@ async fn main() { ...@@ -595,7 +610,7 @@ async fn main() {
.map_err(|e| { .map_err(|e| {
log::error!(target: LOG_TARGET, "DryRun error: {:?}", 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| { .map_err(|e| {
log::error!(target: LOG_TARGET, "EmergencySolution error: {:?}", e); log::error!(target: LOG_TARGET, "EmergencySolution error: {:?}", e);
}), }),
......
...@@ -21,7 +21,7 @@ use crate::{ ...@@ -21,7 +21,7 @@ use crate::{
}; };
use codec::Encode; use codec::Encode;
use jsonrpsee_ws_client::{ use jsonrpsee_ws_client::{
types::{traits::SubscriptionClient, v2::params::JsonRpcParams, Subscription}, types::{traits::SubscriptionClient, Subscription},
WsClient, WsClient,
}; };
use sc_transaction_pool_api::TransactionStatus; use sc_transaction_pool_api::TransactionStatus;
...@@ -71,106 +71,113 @@ macro_rules! monitor_cmd_for { ($runtime:tt) => { paste::paste! { ...@@ -71,106 +71,113 @@ macro_rules! monitor_cmd_for { ($runtime:tt) => { paste::paste! {
signer: Signer, signer: Signer,
) -> Result<(), Error<$crate::[<$runtime _runtime_exports>]::Runtime>> { ) -> Result<(), Error<$crate::[<$runtime _runtime_exports>]::Runtime>> {
use $crate::[<$runtime _runtime_exports>]::*; use $crate::[<$runtime _runtime_exports>]::*;
let (sub, unsub) = if config.listen == "head" { let (sub, unsub) = if config.listen == "head" {
("chain_subscribeNewHeads", "chain_unsubscribeNewHeads") ("chain_subscribeNewHeads", "chain_unsubscribeNewHeads")
} else { } else {
("chain_subscribeFinalizedHeads", "chain_unsubscribeFinalizedHeads") ("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();
while let Some(now) = subscription.next().await? {
let hash = now.hash();
log::debug!(target: LOG_TARGET, "new event at #{:?} ({:?})", now.number, hash);
// if the runtime version has changed, terminate
crate::check_versions::<Runtime>(client, false).await?;
// we prefer doing this check before fetching anything into a remote-ext.
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.");
continue;
};
// NOTE: we don't check the score of any of the submitted solutions. If we submit a weak
// one, as long as we are valid, we will end up getting our deposit back, so not a big
// deal for now. Note that to avoid an unfeasible solution, we should make sure that we
// only start the process on a finalized snapshot. If the signed phase is long enough,
// this will not be a solution.
// grab an externalities without staking, just the election snapshot.
let mut ext = crate::create_election_ext::<Runtime, Block>(shared.uri.clone(), Some(hash), vec![]).await?;
if ensure_no_previous_solution::<Runtime, Block>(&mut ext, &signer.account).await.is_err() {
log::debug!(target: LOG_TARGET, "We already have a solution in this phase, skipping.");
continue;
}
let (raw_solution, witness) = crate::mine_with::<Runtime>(&config.solver, &mut ext)?;
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 .await
{ .unwrap();
Ok(sub) => sub,
Err(why) => { while let Some(now) = subscription.next().await? {
// This usually happens when we've been busy with mining for a few blocks, and now we're receiving the let hash = now.hash();
// subscriptions of blocks in which we were busy. In these blocks, we still don't have a solution, so we log::trace!(target: LOG_TARGET, "new event at #{:?} ({:?})", now.number, hash);
// 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, // if the runtime version has changed, terminate.
// we should make this multi-threaded and do the computation outside of this callback. crate::check_versions::<Runtime>(client).await?;
log::warn!(target: LOG_TARGET, "failing to submit a transaction {:?}. continuing...", why);
continue // we prefer doing this check before fetching anything into a remote-ext.
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.");
continue;
};
// grab an externalities without staking, just the election snapshot.
let mut ext = crate::create_election_ext::<Runtime, Block>(
shared.uri.clone(),
Some(hash),
vec![],
).await?;
if ensure_no_previous_solution::<Runtime, Block>(&mut ext, &signer.account).await.is_err() {
log::debug!(target: LOG_TARGET, "We already have a solution in this phase, skipping.");
continue;
} }
};
// mine a solution, and run feasibility check on it as well.
let _success = while let Some(status_update) = tx_subscription.next().await? { let (raw_solution, witness) = crate::mine_with::<Runtime>(&config.solver, &mut ext, true)?;
log::trace!(target: LOG_TARGET, "status update {:?}", status_update); log::info!(target: LOG_TARGET, "mined solution with {:?}", &raw_solution.score);
match status_update {
TransactionStatus::Ready | TransactionStatus::Broadcast(_) | TransactionStatus::Future => continue, let nonce = crate::get_account_info::<Runtime>(client, &signer.account, Some(hash))
TransactionStatus::InBlock(hash) => { .await?
log::info!(target: LOG_TARGET, "included at {:?}", hash); .map(|i| i.nonce)
let key = frame_support::storage::storage_prefix(b"System", b"Events"); .expect(crate::signer::SIGNER_ACCOUNT_WILL_EXIST);
let events = get_storage::<Vec<frame_system::EventRecord<Event, <Block as BlockT>::Hash>>, let tip = 0 as Balance;
>(client, params!{ key, hash }).await?.unwrap_or_default(); let period = <Runtime as frame_system::Config>::BlockHashCount::get() / 2;
log::info!(target: LOG_TARGET, "events at inclusion {:?}", events); let current_block = now.number.saturating_sub(1);
} let era = sp_runtime::generic::Era::mortal(period.into(), current_block.into());
TransactionStatus::Retracted(hash) => { log::trace!(
log::info!(target: LOG_TARGET, "Retracted at {:?}", hash); target: LOG_TARGET, "transaction mortality: {:?} -> {:?}",
} era.birth(current_block.into()),
TransactionStatus::Finalized(hash) => { era.death(current_block.into()),
log::info!(target: LOG_TARGET, "Finalized at {:?}", hash); );
break 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
} }
_ => { };
log::warn!(target: LOG_TARGET, "Stopping listen due to other status {:?}", status_update);
break 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(()) log::warn!(target: LOG_TARGET, "subscription to {} terminated. Retrying..", sub)
}
} }
}}} }}}
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment