Clear mempool at a network upgrade

zebra-state/src/lib.rs

@@ -35,7 +35,10 @@ pub use constants::MAX_BLOCK_REORG_HEIGHT;
 pub use error::{BoxError, CloneError, CommitBlockError, ValidateContextError};
 pub use request::{FinalizedBlock, HashOrHeight, PreparedBlock, Request};
 pub use response::Response;
-pub use service::{chain_tip::LatestChainTip, init};
+pub use service::{
+    chain_tip::{ChainTipChange, LatestChainTip, TipAction},
+    init,
+};
 
 #[cfg(any(test, feature = "proptest-impl"))]
 pub use service::init_test;

zebrad/src/commands/start.rs

@@ -55,7 +55,7 @@ impl StartCmd {
 
         info!("initializing node state");
         // TODO: use ChainTipChange to get tip changes (#2374, #2710, #2711, #2712, #2713, #2714)
-        let (state_service, latest_chain_tip, _chain_tip_change) =
+        let (state_service, latest_chain_tip, chain_tip_change) =
             zebra_state::init(config.state.clone(), config.network.network);
         let state = ServiceBuilder::new().buffer(20).service(state_service);
 
@@ -96,6 +96,7 @@ impl StartCmd {
             state,
             tx_verifier,
             sync_status.clone(),
+            chain_tip_change,
         ));
         let mempool = ServiceBuilder::new().buffer(20).service(mempool_service);
 

zebrad/src/components/inbound/tests.rs

@@ -24,6 +24,8 @@ async fn mempool_requests_for_transactions() {
     let address_book = AddressBook::new(SocketAddr::from_str("0.0.0.0:0").unwrap(), Span::none());
     let address_book = Arc::new(std::sync::Mutex::new(address_book));
     let (sync_status, _recent_syncs) = SyncStatus::new();
+    let (_state_service, _latest_chain_tip, chain_tip_change) =
+        zebra_state::init(state_config.clone(), network);
 
     let (state, _, _) = zebra_state::init(state_config, network);
     let state_service = ServiceBuilder::new().buffer(1).service(state);
@@ -38,6 +40,7 @@ async fn mempool_requests_for_transactions() {
         state_service.clone(),
         transaction_verifier,
         sync_status,
+        chain_tip_change,
     );
 
     let added_transactions = add_some_stuff_to_mempool(&mut mempool_service, network);

zebrad/src/components/mempool.rs

@@ -17,6 +17,7 @@ use zebra_chain::{
 use zebra_consensus::{error::TransactionError, transaction};
 use zebra_network as zn;
 use zebra_state as zs;
+use zebra_state::{ChainTipChange, TipAction};
 
 pub use crate::BoxError;
 
@@ -82,6 +83,8 @@ pub struct Mempool {
     /// Allows checking if we are near the tip to enable/disable the mempool.
     #[allow(dead_code)]
     sync_status: SyncStatus,
+
+    chain_tip_change: ChainTipChange,
 }
 
 impl Mempool {
@@ -92,20 +95,23 @@ impl Mempool {
         state: State,
         tx_verifier: TxVerifier,
         sync_status: SyncStatus,
+        chain_tip_change: ChainTipChange,
     ) -> Self {
         let tx_downloads = Box::pin(TxDownloads::new(
             Timeout::new(outbound, TRANSACTION_DOWNLOAD_TIMEOUT),
             Timeout::new(tx_verifier, TRANSACTION_VERIFY_TIMEOUT),
             state,
         ));
+
         Mempool {
             storage: Default::default(),
             tx_downloads,
             sync_status,
+            chain_tip_change,
         }
     }
 
-    ///  Get the storage field of the mempool for testing purposes.
+    /// Get the storage field of the mempool for testing purposes.
     #[cfg(test)]
     pub fn storage(&mut self) -> &mut storage::Storage {
         &mut self.storage
@@ -134,6 +140,13 @@ impl Service<Request> for Mempool {
         Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        // Clear the mempool if there has been a chain tip reset.
+        if let Some(Ok(TipAction::Reset { height: _, hash: _ })) =
+            self.chain_tip_change.tip_change().now_or_never()
+        {
+            self.storage.clear();
+        }
+
         // Clean up completed download tasks and add to mempool if successful
         while let Poll::Ready(Some(r)) = self.tx_downloads.as_mut().poll_next(cx) {
             if let Ok(tx) = r {

zebrad/src/components/mempool/storage.rs

@@ -22,7 +22,8 @@ pub enum State {
     /// no longer belongs in the mempool.
     Confirmed(block::Hash),
     /// Stayed in mempool for too long without being mined.
-    // TODO(2021-08-20): set expiration at 2 weeks? This is what Bitcoin does.
+    // TODO(2021-09-09): Implement ZIP-203: Validate Transaction Expiry Height.
+    // TODO(2021-09-09): https://github.com/ZcashFoundation/zebra/issues/2387
     Expired,
     /// Transaction fee is too low for the current mempool state.
     LowFee,
@@ -139,4 +140,10 @@ impl Storage {
             .filter(|tx| self.rejected.contains_key(tx))
             .collect()
     }
+
+    /// Clears the whole mempool storage.
+    pub fn clear(&mut self) {
+        self.verified.clear();
+        self.rejected.clear();
+    }
 }

zebrad/src/components/mempool/tests.rs

@@ -17,6 +17,8 @@ async fn mempool_service_basic() -> Result<(), Report> {
     let state_config = StateConfig::ephemeral();
     let (peer_set, _) = mock_peer_set();
     let (sync_status, _recent_syncs) = SyncStatus::new();
+    let (_state_service, _latest_chain_tip, chain_tip_change) =
+        zebra_state::init(state_config.clone(), network);
 
     let (state, _, _) = zebra_state::init(state_config, network);
     let state_service = ServiceBuilder::new().buffer(1).service(state);
@@ -33,6 +35,7 @@ async fn mempool_service_basic() -> Result<(), Report> {
         state_service.clone(),
         tx_verifier,
         sync_status,
+        chain_tip_change,
     );
     // Insert the genesis block coinbase transaction into the mempool storage.
     service.storage.insert(genesis_transactions.1[0].clone())?;