feat: persist ExpectedUtxo to disk (db)

closes #172, #137.

fixes loss-of-funds scenario.

changes:
 * add RustyWalletDatabase::expected_utxos (DbtVec)
 * remove UtxoNotificationPool
 * remove UtxoNotifier::PeerUnsigned
 * add WalletState methods:
    add_expected_utxo(), scan_for_expected_utxos()
 * disables timer call to prune_stale_expected_utxos() for now.
 + remove cli args:
    max_utxo_notification_size,
    max_unconfirmed_utxo_notification_count_per_peer
 * mod utxo_notification_pool --> expected_utxo
 * derive Hash for Timestamp

tests:
 * adapt existing tests to changes
 * move tests from utxo_notification_pool into wallet_state
 * adds regression tests for issue #172 that verify:
    1.  expected_utxo are persisted if db is written to disk.
    2.  expected_utxo are not persisted if db is not written to disk.

src/config_models/cli_args.rs

@@ -66,23 +66,6 @@ pub struct Args {
     #[clap(long, default_value = "1G", value_name = "SIZE")]
     pub max_mempool_size: ByteSize,
 
-    /// Prune the pool of UTXO notification when it exceeds this size in RAM.
-    ///
-    /// Units: B (bytes), K (kilobytes), M (megabytes), G (gigabytes)
-    ///
-    /// E.g. --max-utxo-notification-size 50M
-    #[clap(long, default_value = "50M", value_name = "SIZE")]
-    pub max_utxo_notification_size: ByteSize,
-
-    /// Maximum number of unconfirmed expected UTXOs that can be stored for each peer.
-    ///
-    /// You may want to increase this number from its default value if
-    /// you're running a node that receives a very high number of UTXOs.
-    ///
-    /// E.g. --max_unconfirmed_utxo_notification_count_per_peer 5000
-    #[clap(long, default_value = "1000", value_name = "COUNT")]
-    pub max_unconfirmed_utxo_notification_count_per_peer: usize,
-
     /// Port on which to listen for peer connections.
     #[clap(long, default_value = "9798", value_name = "PORT")]
     pub peer_port: u16,

src/main_loop.rs

@@ -44,7 +44,7 @@ const PEER_DISCOVERY_INTERVAL_IN_SECONDS: u64 = 120;
 const SYNC_REQUEST_INTERVAL_IN_SECONDS: u64 = 3;
 const MEMPOOL_PRUNE_INTERVAL_IN_SECS: u64 = 30 * 60; // 30mins
 const MP_RESYNC_INTERVAL_IN_SECS: u64 = 59;
-const UTXO_NOTIFICATION_POOL_PRUNE_INTERVAL_IN_SECS: u64 = 19 * 60; // 19 mins
+const EXPECTED_UTXOS_PRUNE_INTERVAL_IN_SECS: u64 = 19 * 60; // 19 mins
 
 const SANCTION_PEER_TIMEOUT_FACTOR: u64 = 40;
 const POTENTIAL_PEER_MAX_COUNT_AS_A_FACTOR_OF_MAX_PEERS: usize = 20;
@@ -821,7 +821,7 @@ impl MainLoopHandler {
 
         // Set removal of stale notifications for incoming UTXOs
         let utxo_notification_cleanup_timer_interval =
-            Duration::from_secs(UTXO_NOTIFICATION_POOL_PRUNE_INTERVAL_IN_SECS);
+            Duration::from_secs(EXPECTED_UTXOS_PRUNE_INTERVAL_IN_SECS);
         let utxo_notification_cleanup_timer = time::sleep(utxo_notification_cleanup_timer_interval);
         tokio::pin!(utxo_notification_cleanup_timer);
 
@@ -979,7 +979,15 @@ impl MainLoopHandler {
                 // Handle incoming UTXO notification cleanup, i.e. removing stale/too old UTXO notification from pool
                 _ = &mut utxo_notification_cleanup_timer => {
                     debug!("Timer: UTXO notification pool cleanup job");
-                    self.global_state_lock.lock_mut(|s| s.wallet_state.expected_utxos.prune_stale_utxo_notifications()).await;
+
+                    // Danger: possible loss of funds.
+                    //
+                    // See description of prune_stale_expected_utxos().
+                    //
+                    // This call is disabled until such time as a thorough
+                    // evaluation and perhaps reimplementation determines that
+                    // it can be called safely without possible loss of funds.
+                    // self.global_state_lock.lock_mut(|s| s.wallet_state.prune_stale_expected_utxos()).await;
 
                     utxo_notification_cleanup_timer.as_mut().reset(tokio::time::Instant::now() + utxo_notification_cleanup_timer_interval);
                 }

src/mine_loop.rs

@@ -39,8 +39,8 @@ use crate::models::blockchain::type_scripts::TypeScript;
 use crate::models::channel::*;
 use crate::models::consensus::timestamp::Timestamp;
 use crate::models::shared::SIZE_20MB_IN_BYTES;
-use crate::models::state::wallet::utxo_notification_pool::ExpectedUtxo;
-use crate::models::state::wallet::utxo_notification_pool::UtxoNotifier;
+use crate::models::state::wallet::expected_utxo::ExpectedUtxo;
+use crate::models::state::wallet::expected_utxo::UtxoNotifier;
 use crate::models::state::wallet::WalletSecret;
 use crate::models::state::GlobalState;
 use crate::models::state::GlobalStateLock;

src/models/blockchain/transaction/mod.rs

@@ -47,7 +47,7 @@ use crate::models::blockchain::block::mutator_set_update::MutatorSetUpdate;
 use crate::models::consensus::mast_hash::MastHash;
 use crate::models::consensus::ValidityTree;
 use crate::models::consensus::WitnessType;
-use crate::models::state::wallet::utxo_notification_pool::ExpectedUtxo;
+use crate::models::state::wallet::expected_utxo::ExpectedUtxo;
 use crate::prelude::triton_vm;
 use crate::prelude::twenty_first;
 use crate::util_types::mutator_set::addition_record::AdditionRecord;

src/models/blockchain/transaction/transaction_output.rs

@@ -6,8 +6,8 @@ use crate::models::blockchain::transaction::PublicAnnouncement;
 use crate::models::blockchain::type_scripts::neptune_coins::NeptuneCoins;
 use crate::models::state::wallet::address::ReceivingAddress;
 use crate::models::state::wallet::address::SpendingKey;
-use crate::models::state::wallet::utxo_notification_pool::ExpectedUtxo;
-use crate::models::state::wallet::utxo_notification_pool::UtxoNotifier;
+use crate::models::state::wallet::expected_utxo::ExpectedUtxo;
+use crate::models::state::wallet::expected_utxo::UtxoNotifier;
 use crate::models::state::wallet::wallet_state::WalletState;
 use crate::prelude::twenty_first::math::digest::Digest;
 use crate::prelude::twenty_first::util_types::algebraic_hasher::AlgebraicHasher;
@@ -277,7 +277,7 @@ impl From<&TxOutputList> for Vec<Utxo> {
 
 impl From<&TxOutputList> for Vec<ExpectedUtxo> {
     fn from(list: &TxOutputList) -> Self {
-        list.expected_utxos_iter().into_iter().collect()
+        list.expected_utxos_iter().collect()
     }
 }
 
@@ -330,7 +330,7 @@ impl TxOutputList {
     }
 
     /// retrieves expected_utxos from possible sub-set of the list
-    pub fn expected_utxos_iter(&self) -> impl IntoIterator<Item = ExpectedUtxo> + '_ {
+    pub fn expected_utxos_iter(&self) -> impl Iterator<Item = ExpectedUtxo> + '_ {
         self.0.iter().filter_map(|u| match &u.utxo_notification {
             UtxoNotification::OffChain(eu) => Some(*eu.clone()),
             _ => None,
@@ -346,7 +346,7 @@ impl TxOutputList {
 
     /// retrieves expected_utxos from possible sub-set of the list
     pub fn expected_utxos(&self) -> Vec<ExpectedUtxo> {
-        self.expected_utxos_iter().into_iter().collect()
+        self.expected_utxos_iter().collect()
     }
 }
 

src/models/channel.rs

@@ -10,7 +10,7 @@ use super::blockchain::block::block_height::BlockHeight;
 use super::blockchain::block::Block;
 use super::blockchain::transaction::Transaction;
 use super::peer::TransactionNotification;
-use super::state::wallet::utxo_notification_pool::ExpectedUtxo;
+use super::state::wallet::expected_utxo::ExpectedUtxo;
 
 #[derive(Clone, Debug)]
 pub enum MainToMiner {

src/models/consensus/timestamp.rs

@@ -24,7 +24,7 @@ use tasm_lib::twenty_first::math::bfield_codec::BFieldCodec;
 /// milliseconds elapsed since the Unix epoch (00:00 UTC on 1 Jan 1970) using
 /// a single BFieldElement.
 #[derive(
-    Debug, Clone, Copy, BFieldCodec, PartialEq, Eq, Serialize, Deserialize, GetSize, Default,
+    Debug, Clone, Copy, Hash, BFieldCodec, PartialEq, Eq, Serialize, Deserialize, GetSize, Default,
 )]
 pub struct Timestamp(pub BFieldElement);
 

src/models/state/archival_state.rs

@@ -861,7 +861,8 @@ mod archival_state_tests {
     use crate::models::blockchain::type_scripts::neptune_coins::NeptuneCoins;
     use crate::models::consensus::timestamp::Timestamp;
     use crate::models::state::archival_state::ArchivalState;
-    use crate::models::state::wallet::utxo_notification_pool::UtxoNotifier;
+    use crate::models::state::wallet::expected_utxo::ExpectedUtxo;
+    use crate::models::state::wallet::expected_utxo::UtxoNotifier;
     use crate::models::state::wallet::WalletSecret;
     use crate::tests::shared::add_block_to_archival_state;
     use crate::tests::shared::make_mock_block_with_valid_pow;
@@ -1587,28 +1588,26 @@ mod archival_state_tests {
             let mut genesis_state = genesis_state_lock.lock_guard_mut().await;
             genesis_state
                 .wallet_state
-                .expected_utxos
-                .add_expected_utxo(
+                .add_expected_utxo(ExpectedUtxo::new(
                     cb_utxo,
                     cb_output_randomness,
                     genesis_spending_key.privacy_preimage,
                     UtxoNotifier::OwnMiner,
-                )
-                .unwrap();
+                ))
+                .await;
         }
         {
             let mut alice_state = alice_state_lock.lock_guard_mut().await;
             for rec_data in tx_outputs_for_alice {
                 alice_state
                     .wallet_state
-                    .expected_utxos
-                    .add_expected_utxo(
+                    .add_expected_utxo(ExpectedUtxo::new(
                         rec_data.utxo.clone(),
                         rec_data.sender_randomness,
                         alice_spending_key.privacy_preimage,
                         UtxoNotifier::Cli,
-                    )
-                    .unwrap();
+                    ))
+                    .await;
             }
         }
 
@@ -1617,14 +1616,13 @@ mod archival_state_tests {
             for rec_data in tx_outputs_for_bob {
                 bob_state
                     .wallet_state
-                    .expected_utxos
-                    .add_expected_utxo(
+                    .add_expected_utxo(ExpectedUtxo::new(
                         rec_data.utxo.clone(),
                         rec_data.sender_randomness,
                         bob_spending_key.privacy_preimage,
                         UtxoNotifier::Cli,
-                    )
-                    .unwrap();
+                    ))
+                    .await;
             }
         }
 
@@ -1783,43 +1781,40 @@ mod archival_state_tests {
                 .lock_guard_mut()
                 .await
                 .wallet_state
-                .expected_utxos
-                .add_expected_utxo(
+                .add_expected_utxo(ExpectedUtxo::new(
                     rec_data.utxo.clone(),
                     rec_data.sender_randomness,
                     genesis_spending_key.privacy_preimage,
                     UtxoNotifier::Cli,
-                )
-                .unwrap();
+                ))
+                .await;
         }
 
         for rec_data in tx_outputs_from_bob {
             genesis_state_lock
                 .lock_guard_mut()
                 .await
                 .wallet_state
-                .expected_utxos
-                .add_expected_utxo(
+                .add_expected_utxo(ExpectedUtxo::new(
                     rec_data.utxo.clone(),
                     rec_data.sender_randomness,
                     genesis_spending_key.privacy_preimage,
                     UtxoNotifier::Cli,
-                )
-                .unwrap();
+                ))
+                .await;
         }
 
         genesis_state_lock
             .lock_guard_mut()
             .await
             .wallet_state
-            .expected_utxos
-            .add_expected_utxo(
+            .add_expected_utxo(ExpectedUtxo::new(
                 cb_utxo_block_2,
                 cb_sender_randomness_block_2,
                 genesis_spending_key.privacy_preimage,
                 UtxoNotifier::Cli,
-            )
-            .unwrap();
+            ))
+            .await;
 
         // Update chain states
         for state_lock in [&genesis_state_lock, &alice_state_lock, &bob_state_lock] {

src/models/state/mempool.rs

@@ -456,7 +456,8 @@ mod tests {
     use crate::models::blockchain::transaction::TxOutput;
     use crate::models::blockchain::type_scripts::neptune_coins::NeptuneCoins;
     use crate::models::shared::SIZE_20MB_IN_BYTES;
-    use crate::models::state::wallet::utxo_notification_pool::UtxoNotifier;
+    use crate::models::state::wallet::expected_utxo::ExpectedUtxo;
+    use crate::models::state::wallet::expected_utxo::UtxoNotifier;
     use crate::models::state::wallet::WalletSecret;
     use crate::tests::shared::make_mock_block;
     use crate::tests::shared::make_mock_transaction_with_wallet;
@@ -644,14 +645,13 @@ mod tests {
         // Update both states with block 1
         other_global_state
             .wallet_state
-            .expected_utxos
-            .add_expected_utxo(
+            .add_expected_utxo(ExpectedUtxo::new(
                 coinbase_utxo_1,
                 cb_sender_randomness_1,
                 other_receiver_spending_key.privacy_preimage,
                 UtxoNotifier::OwnMiner,
-            )
-            .expect("UTXO notification from miner must be accepted");
+            ))
+            .await;
         other_global_state
             .set_new_tip(block_1.clone())
             .await

src/models/state/mod.rs

@@ -24,7 +24,7 @@ use twenty_first::math::digest::Digest;
 use twenty_first::util_types::algebraic_hasher::AlgebraicHasher;
 use wallet::address::ReceivingAddress;
 use wallet::address::SpendingKey;
-use wallet::utxo_notification_pool::UtxoNotifier;
+use wallet::expected_utxo::UtxoNotifier;
 use wallet::wallet_state::WalletState;
 use wallet::wallet_status::WalletStatus;
 
@@ -36,7 +36,7 @@ use crate::locks::tokio as sync_tokio;
 use crate::models::blockchain::transaction::UtxoNotifyMethod;
 use crate::models::peer::HandshakeData;
 use crate::models::state::wallet::monitored_utxo::MonitoredUtxo;
-use crate::models::state::wallet::utxo_notification_pool::ExpectedUtxo;
+use crate::models::state::wallet::expected_utxo::ExpectedUtxo;
 use crate::prelude::twenty_first;
 use crate::time_fn_call_async;
 use crate::util_types::mutator_set::mutator_set_accumulator::MutatorSetAccumulator;
@@ -814,12 +814,14 @@ impl GlobalState {
         expected_utxos: impl IntoIterator<Item = ExpectedUtxo>,
     ) -> Result<()> {
         for expected_utxo in expected_utxos.into_iter() {
-            self.wallet_state.expected_utxos.add_expected_utxo(
-                expected_utxo.utxo,
-                expected_utxo.sender_randomness,
-                expected_utxo.receiver_preimage,
-                expected_utxo.received_from,
-            )?;
+            self.wallet_state
+                .add_expected_utxo(ExpectedUtxo::new(
+                    expected_utxo.utxo,
+                    expected_utxo.sender_randomness,
+                    expected_utxo.receiver_preimage,
+                    expected_utxo.received_from,
+                ))
+                .await;
         }
         Ok(())
     }
@@ -1291,14 +1293,13 @@ impl GlobalState {
                 // Notify wallet to expect the coinbase UTXO, as we mined this block
                 myself
                     .wallet_state
-                    .expected_utxos
-                    .add_expected_utxo(
+                    .add_expected_utxo(ExpectedUtxo::new(
                         coinbase_info.utxo,
                         coinbase_info.sender_randomness,
                         coinbase_info.receiver_preimage,
                         UtxoNotifier::OwnMiner,
-                    )
-                    .expect("UTXO notification from miner must be accepted");
+                    ))
+                    .await;
             }
 
             // Get parent of tip for mutator-set data needed for various updates. Parent of the
@@ -1398,7 +1399,7 @@ mod global_state_tests {
 
     use crate::config_models::network::Network;
     use crate::models::blockchain::block::Block;
-    use crate::models::state::wallet::utxo_notification_pool::UtxoNotifier;
+    use crate::models::state::wallet::expected_utxo::UtxoNotifier;
     use crate::tests::shared::add_block_to_light_state;
     use crate::tests::shared::make_mock_block;
     use crate::tests::shared::make_mock_block_with_valid_pow;
@@ -2111,29 +2112,27 @@ mod global_state_tests {
                 .lock_guard_mut()
                 .await
                 .wallet_state
-                .expected_utxos
-                .add_expected_utxo(
+                .add_expected_utxo(ExpectedUtxo::new(
                     rec_data.utxo.clone(),
                     rec_data.sender_randomness,
                     alice_spending_key.privacy_preimage,
                     UtxoNotifier::Cli,
-                )
-                .unwrap();
+                ))
+                .await;
         }
 
         for rec_data in tx_outputs_for_bob {
             bob_state_lock
                 .lock_guard_mut()
                 .await
                 .wallet_state
-                .expected_utxos
-                .add_expected_utxo(
+                .add_expected_utxo(ExpectedUtxo::new(
                     rec_data.utxo.clone(),
                     rec_data.sender_randomness,
                     bob_spending_key.privacy_preimage,
                     UtxoNotifier::Cli,
-                )
-                .unwrap();
+                ))
+                .await;
         }
 
         genesis_state_lock