Merge pull request #2057 from mintlayer/wallet_generate_blocks_fix_mempool_recoverable_error

wallet: `Controller::generate_blocks` now retries on "recoverable mempool errors"

Author: ImplOfAnImpl

Cargo.lock

@@ -47,62 +47,93 @@ pub use detail::timestamp_searcher::{TimestampSearchData, find_timestamps_for_st
 pub enum BlockProductionError {
     #[error("Failed to retrieve chainstate info")]
     ChainstateInfoRetrievalError,
+
     #[error("Wait for chainstate to sync before producing blocks")]
     ChainstateWaitForSync,
+
     #[error("Subsystem call error")]
     SubsystemCallError(#[from] CallError),
+
     #[error("Failed to add transaction {0}: {1}")]
     FailedToAddTransaction(Id<Transaction>, TxAccumulatorError),
+
     #[error("Block creation error: {0}")]
     FailedToConstructBlock(#[from] BlockCreationError),
+
     #[error("Initialization of consensus failed: {0}")]
     FailedConsensusInitialization(#[from] ConsensusCreationError),
+
     #[error("Block production cancelled")]
     Cancelled,
+
     #[error("Failed to retrieve peer count: {0}")]
     PeerCountRetrievalError(String),
+
     #[error("Connected peers {0} is below the required peer threshold {0}")]
     PeerCountBelowRequiredThreshold(usize, usize),
+
     #[error("Block not found in this round")]
     TryAgainLater,
+
     #[error("Job already exists")]
     JobAlreadyExists(JobKey),
+
     #[error("Job manager error: {0}")]
     JobManagerError(#[from] JobManagerError),
+
     #[error("Mempool failed to construct block: {0}")]
     MempoolBlockConstruction(#[from] mempool::error::BlockConstructionError),
+
     #[error("Failed to decrypt generate-block input data: {0}")]
     E2eError(#[from] ephemeral_e2e::error::Error),
+
     #[error("Overflowed when calculating a block timestamp: {0} + {1}")]
     TimestampOverflow(BlockTimestamp, u64),
+
     #[error("Chainstate error: `{0}`")]
     ChainstateError(#[from] consensus::ChainstateError),
+
     #[error("Wrong height range: {0}, {1}")]
     WrongHeightRange(BlockHeight, BlockHeight),
+
     #[error("Block at height {0} doesn't exist")]
     NoBlockForHeight(BlockHeight),
+
     #[error("Block index missing for block {0}")]
     InconsistentDbMissingBlockIndex(Id<GenBlock>),
+
     #[error("Unexpected consensus type: None")]
     UnexpectedConsensusTypeNone,
+
     #[error("Unexpected consensus type: PoW")]
     UnexpectedConsensusTypePoW,
+
     #[error("Pool data for pool {0} not found")]
     PoolDataNotFound(PoolId),
+
     #[error("Balance for pool {0} not found")]
     PoolBalanceNotFound(PoolId),
+
     #[error("PoS accounting error: {0}")]
     PoSAccountingError(#[from] detail::utils::PoSAccountingError),
+
     #[error("PoS data provided when consensus is supposed to be ignored")]
     PoSInputDataProvidedWhenIgnoringConsensus,
+
     #[error("PoW data provided when consensus is supposed to be ignored")]
     PoWInputDataProvidedWhenIgnoringConsensus,
-    #[error("Recoverable mempool error")]
+
+    // Note: the string representation of this error is checked on the client side of node RPC,
+    // this is why it was put into a separate constant.
+    #[error("{RECOVERABLE_MEMPOOL_ERROR_MSG}")]
     RecoverableMempoolError,
+
     #[error("Task exited prematurely")]
     TaskExitedPrematurely,
 }
 
+pub const RECOVERABLE_MEMPOOL_ERROR_MSG: &str = "Blockprod recoverable mempool error";
+
 pub type BlockProductionSubsystem = Box<dyn BlockProductionInterface>;
 pub type BlockProductionHandle = subsystem::Handle<dyn BlockProductionInterface>;
 

blockprod/src/lib.rs

@@ -19,7 +19,7 @@ use thiserror::Error;
 
 use chainstate::{ChainstateError, tx_verifier::error::ConnectTransactionError};
 use common::{
-    chain::{Block, GenBlock, Transaction},
+    chain::{Block, Transaction},
     primitives::{H256, Id, amount::DisplayAmount},
 };
 
@@ -33,9 +33,6 @@ pub enum BlockConstructionError {
     #[error(transparent)]
     Validity(#[from] TxValidationError),
 
-    #[error("The tip moved during block construction: {0:?} -> {1:?}")]
-    TipMoved(Id<GenBlock>, Id<GenBlock>),
-
     #[error("Subsystem call error: {0}")]
     SubsystemCallError(#[from] subsystem::error::CallError),
 

mempool/src/error/mod.rs

@@ -247,12 +247,16 @@ pub fn collect_txs<M>(
 
     let final_chainstate_tip =
         utxo::UtxosView::best_block_hash(&chainstate).expect("cannot fetch tip");
-    ensure!(
-        mempool_tip == final_chainstate_tip,
-        BlockConstructionError::TipMoved(mempool_tip, final_chainstate_tip),
-    );
 
-    Ok(Some(tx_accumulator))
+    if mempool_tip == final_chainstate_tip {
+        Ok(Some(tx_accumulator))
+    } else {
+        // The tip has moved, so return "Ok(None)" to signal a recoverable error.
+        // TODO: perhaps the chainstate tip check is redundant here, because the tip change can't
+        // have affected the mempool at this point, so all collected txs are valid for inclusion
+        // in a block that has `tx_accumulator.expected_tip()` as its parent.
+        Ok(None)
+    }
 }
 
 /// Return at most `tx_count` tx ids from `tx_ids`, ordering them by score and ancestry

mempool/src/pool/tx_pool/collect_txs.rs

@@ -73,7 +73,7 @@ def create_blocks(self, blocks_count: int, initial_transfer_amount_atoms: int) -
         return block_ids
 
     # Need to call this function after the tip has changed, if a new block is to be generated
-    # afterwards. Otherwise the block generation may fail with "Recoverable mempool error".
+    # afterwards. Otherwise the block generation may fail with "Blockprod recoverable mempool error".
     def wait_for_mempool_update(self, tip_id: str):
         node = self.nodes[0]
         self.wait_until(lambda: node.mempool_local_best_block_id() == tip_id, timeout=5)

test/functional/node_bootstrapping.py

@@ -38,6 +38,7 @@ use helpers::{
     fetch_input_infos, fetch_rpc_token_info, fetch_utxo, fetch_utxo_extra_info, into_balances,
 };
 use itertools::Itertools as _;
+use node_comm::node_traits::NodeInterfaceError as _;
 use runtime_wallet::RuntimeWallet;
 use std::{
     collections::{BTreeMap, BTreeSet},
@@ -773,26 +774,116 @@ where
 
     /// Try to generate the `block_count` number of blocks.
     /// The function may return an error early if some attempt fails.
+    ///
+    /// Note that this function is intended to be used on regtest/signet to populate the chainstate
+    /// and it won't work reliably if a large number of blocks enters the chainstate via other means
+    /// at the same time (e.g. via p2p during initial block download).
     pub async fn generate_blocks(
         &mut self,
         account_index: U31,
         block_count: u32,
     ) -> Result<(), ControllerError<N>> {
-        for _ in 0..block_count {
-            self.sync_once().await?;
-            let block = self
-                .generate_block(
-                    account_index,
-                    vec![],
-                    vec![],
-                    PackingStrategy::FillSpaceFromMempool,
-                )
-                .await?;
+        let mut recoverable_errors_seen_count = 0;
+
+        for block_idx in 0..block_count {
+            // Perform a few attempts to produce a block, retrying on a "recoverable mempool error",
+            // which indicates that the block production was aborted because the tip has changed
+            // when collecting transactions from the mempool.
+            // This may happen either because the mempool is lagging behind chainstate (so the
+            // new tip is one of the previously produced blocks) or if blocks enter the chainstate
+            // via other means (e.g. p2p).
+            // Note:
+            // 1) Potentially we may see a "recoverable mempool error" for each of the blocks
+            //    we produce here, so the retry count should be at least "the number of blocks we
+            //    produced so far" minus "the number of recoverable errors seen so far". We add
+            //    a small number to that to have some leeway in case blocks are also entering
+            //    the chainstate via other means.
+            // 2) The alternative to retrying is to wait for a NewTip event from the mempool after
+            //    each block; this would make the function more reliable in the case of the mempool
+            //    lagging, but less reliable in the case when blocks enter chainstate via p2p
+            //    (in which case a newly produced block may not become a tip at all).
+            //    I.e. there seems to be no way to make this function 100% reliable.
+            let recoverable_error_retry_count =
+                (block_idx + 1).saturating_sub(recoverable_errors_seen_count) + 5;
+            let mut recoverable_error_retry_idx = 0;
+            loop {
+                self.sync_once().await?;
+                let block_gen_result = self
+                    .generate_block(
+                        account_index,
+                        vec![],
+                        vec![],
+                        PackingStrategy::FillSpaceFromMempool,
+                    )
+                    .await;
 
-            self.rpc_client
-                .submit_block(block)
-                .await
-                .map_err(ControllerError::NodeCallError)?;
+                match block_gen_result {
+                    Ok(block) => {
+                        self.rpc_client
+                            .submit_block(block)
+                            .await
+                            .map_err(ControllerError::NodeCallError)?;
+                        break;
+                    }
+                    Err(err) => {
+                        let is_recoverable_err = match &err {
+                            ControllerError::NodeCallError(err) => {
+                                err.is_recoverable_mempool_error_during_block_production()
+                            }
+                            ControllerError::SyncError(_)
+                            | ControllerError::NotEnoughBlockHeight(_, _)
+                            | ControllerError::WalletFileError(_, _)
+                            | ControllerError::WalletError(_)
+                            | ControllerError::AddressEncodingError(_)
+                            | ControllerError::NoStakingPool
+                            | ControllerError::FrozenToken(_)
+                            | ControllerError::WalletIsLocked
+                            | ControllerError::StakingRunning
+                            | ControllerError::EndToEndEncryptionError(_)
+                            | ControllerError::NodeNotInSyncYet
+                            | ControllerError::InvalidLookaheadSize
+                            | ControllerError::WalletFileAlreadyOpen
+                            | ControllerError::NoWallet
+                            | ControllerError::SearchForTimestampsFailed(_)
+                            | ControllerError::ExpectingNonEmptyInputs
+                            | ControllerError::ExpectingNonEmptyOutputs
+                            | ControllerError::NoCoinUtxosToPayFeeFrom
+                            | ControllerError::InvalidTxOutput(_)
+                            | ControllerError::NotFungibleToken(_)
+                            | ControllerError::InvalidCoinAmount
+                            | ControllerError::PartiallySignedTransactionError(_)
+                            | ControllerError::InvalidTokenId
+                            | ControllerError::SighashInputCommitmentCreationError(_)
+                            | ControllerError::InvalidHtlcSecretsCount => false,
+                        };
+
+                        if !is_recoverable_err {
+                            return Err(err);
+                        }
+
+                        recoverable_errors_seen_count += 1;
+                        recoverable_error_retry_idx += 1;
+
+                        if recoverable_error_retry_idx >= recoverable_error_retry_count {
+                            log::warn!(
+                                "Too many recoverable mempool errors happened during block production, aborting"
+                            );
+
+                            return Err(err);
+                        }
+
+                        log::info!(
+                            "A recoverable mempool error happened during block production, retrying"
+                        );
+
+                        // Sleep for some amount of time, increasing it as the number of retries grows,
+                        // with the cap of 1 sec.
+                        let delay = Duration::from_millis(100 * recoverable_error_retry_idx as u64);
+                        let delay = std::cmp::min(delay, Duration::from_secs(1));
+                        tokio::time::sleep(delay).await;
+                    }
+                }
+            }
         }
 
         self.sync_once().await

wallet/wallet-controller/src/lib.rs

@@ -25,6 +25,7 @@ wallet-types = { path = "../types", default-features = false }
 anyhow.workspace = true
 async-trait.workspace = true
 base64.workspace = true
+derive_more.workspace = true
 futures.workspace = true
 mockall.workspace = true
 serde_json.workspace = true

wallet/wallet-node-client/Cargo.toml

@@ -48,7 +48,7 @@ use utils::app_version_with_git_info;
 use utils_networking::IpOrSocketAddress;
 use wallet_types::wallet_type::WalletControllerMode;
 
-use crate::node_traits::{MempoolEvents, NodeInterface};
+use crate::node_traits::{MempoolEvents, NodeInterface, NodeInterfaceError};
 
 #[derive(Clone)]
 pub struct WalletHandlesClient {
@@ -107,6 +107,51 @@ impl WalletHandlesClient {
     }
 }
 
+impl NodeInterfaceError for WalletHandlesClientError {
+    fn is_recoverable_mempool_error_during_block_production(&self) -> bool {
+        match self {
+            WalletHandlesClientError::BlockProduction(err) => match err {
+                BlockProductionError::RecoverableMempoolError => true,
+
+                BlockProductionError::ChainstateInfoRetrievalError
+                | BlockProductionError::ChainstateWaitForSync
+                | BlockProductionError::SubsystemCallError(_)
+                | BlockProductionError::FailedToAddTransaction(_, _)
+                | BlockProductionError::FailedToConstructBlock(_)
+                | BlockProductionError::FailedConsensusInitialization(_)
+                | BlockProductionError::Cancelled
+                | BlockProductionError::PeerCountRetrievalError(_)
+                | BlockProductionError::PeerCountBelowRequiredThreshold(_, _)
+                | BlockProductionError::TryAgainLater
+                | BlockProductionError::JobAlreadyExists(_)
+                | BlockProductionError::JobManagerError(_)
+                | BlockProductionError::MempoolBlockConstruction(_)
+                | BlockProductionError::E2eError(_)
+                | BlockProductionError::TimestampOverflow(_, _)
+                | BlockProductionError::ChainstateError(_)
+                | BlockProductionError::WrongHeightRange(_, _)
+                | BlockProductionError::NoBlockForHeight(_)
+                | BlockProductionError::InconsistentDbMissingBlockIndex(_)
+                | BlockProductionError::UnexpectedConsensusTypeNone
+                | BlockProductionError::UnexpectedConsensusTypePoW
+                | BlockProductionError::PoolDataNotFound(_)
+                | BlockProductionError::PoolBalanceNotFound(_)
+                | BlockProductionError::PoSAccountingError(_)
+                | BlockProductionError::PoSInputDataProvidedWhenIgnoringConsensus
+                | BlockProductionError::PoWInputDataProvidedWhenIgnoringConsensus
+                | BlockProductionError::TaskExitedPrematurely => false,
+            },
+
+            WalletHandlesClientError::CallError(_)
+            | WalletHandlesClientError::Chainstate(_)
+            | WalletHandlesClientError::P2p(_)
+            | WalletHandlesClientError::Hex(_)
+            | WalletHandlesClientError::MempoolError(_)
+            | WalletHandlesClientError::AttemptedExit => false,
+        }
+    }
+}
+
 #[async_trait::async_trait]
 impl NodeInterface for WalletHandlesClient {
     type Error = WalletHandlesClientError;

wallet/wallet-node-client/src/handles_client/mod.rs

@@ -39,11 +39,31 @@ use wallet_types::wallet_type::WalletControllerMode;
 
 pub use p2p::{interface::types::ConnectedPeer, types::peer_id::PeerId};
 
-#[mockall::automock(type Error = anyhow::Error;)]
+pub trait NodeInterfaceError: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static {
+    /// Return true if this is the so-called "recoverable mempool error", which may happen
+    /// during block production when the chainstate tip moves and the mempool loses track of
+    /// what transactions are eligible for inclusion in the new block.
+    /// When this happens, the caller code may just retry producing a block.
+    fn is_recoverable_mempool_error_during_block_production(&self) -> bool;
+}
+
+#[derive(Debug, derive_more::Display)]
+#[display("{error}")]
+pub struct MockNodeInterfaceError {
+    pub error: anyhow::Error,
+    pub is_recoverable_mempool_error_during_block_production: bool,
+}
+
+impl NodeInterfaceError for MockNodeInterfaceError {
+    fn is_recoverable_mempool_error_during_block_production(&self) -> bool {
+        self.is_recoverable_mempool_error_during_block_production
+    }
+}
+
+#[mockall::automock(type Error = MockNodeInterfaceError;)]
 #[async_trait::async_trait]
 pub trait NodeInterface {
-    // Note: not requiring the `Error` trait here so that `anyhow::Error` can be used.
-    type Error: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static;
+    type Error: NodeInterfaceError;
 
     async fn is_cold_wallet_node(&self) -> WalletControllerMode;