Merge #306

306: Fix logging and retrying of Monero transaction watching r=thomaseizinger a=thomaseizinger

Hopefully, this should also reduce the load because I am not asking the node every second.

Related: https://github.com/comit-network/xmr-btc-swap/issues/202

Co-authored-by: Thomas Eizinger <thomas@eizinger.io>
Co-authored-by: Daniel Karzel <daniel@comit.network>

monero-rpc/src/rpc/wallet.rs

@@ -471,7 +471,6 @@ struct CheckTxKeyParams {
 #[derive(Clone, Copy, Debug, Deserialize)]
 pub struct CheckTxKey {
     pub confirmations: u32,
-    pub in_pool: bool,
     pub received: u64,
 }
 

swap/src/bin/asb.rs

@@ -24,7 +24,7 @@ use swap::asb::config::{
     initial_setup, query_user_for_initial_testnet_config, read_config, Config, ConfigNotInitialized,
 };
 use swap::database::Database;
-use swap::execution_params::GetExecutionParams;
+use swap::execution_params::{ExecutionParams, GetExecutionParams};
 use swap::fs::default_config_path;
 use swap::monero::Amount;
 use swap::protocol::alice::EventLoop;
@@ -84,6 +84,7 @@ async fn main() -> Result<()> {
                 config.clone(),
                 &wallet_data_dir,
                 seed.derive_extended_private_key(BITCOIN_NETWORK)?,
+                execution_params,
             )
             .await?;
 
@@ -131,6 +132,7 @@ async fn init_wallets(
     config: Config,
     bitcoin_wallet_data_dir: &Path,
     key: impl DerivableKey<Segwitv0> + Clone,
+    execution_params: ExecutionParams,
 ) -> Result<(bitcoin::Wallet, monero::Wallet)> {
     let bitcoin_wallet = bitcoin::Wallet::new(
         config.bitcoin.electrum_rpc_url,
@@ -153,6 +155,7 @@ async fn init_wallets(
         config.monero.wallet_rpc_url.clone(),
         MONERO_NETWORK,
         DEFAULT_WALLET_NAME.to_string(),
+        execution_params.monero_avg_block_time,
     );
 
     // Setup the Monero wallet

swap/src/bin/swap.rs

@@ -23,7 +23,7 @@ use swap::bitcoin::{Amount, TxLock};
 use swap::cli::command::{AliceConnectParams, Arguments, Command, MoneroParams};
 use swap::cli::config::{read_config, Config};
 use swap::database::Database;
-use swap::execution_params::GetExecutionParams;
+use swap::execution_params::{ExecutionParams, GetExecutionParams};
 use swap::network::quote::BidQuote;
 use swap::protocol::bob;
 use swap::protocol::bob::{Builder, EventLoop};
@@ -104,8 +104,13 @@ async fn main() -> Result<()> {
             }
 
             let bitcoin_wallet = init_bitcoin_wallet(bitcoin_network, &config, seed).await?;
-            let (monero_wallet, _process) =
+            let (monero_wallet, _process) = init_monero_wallet(
-                init_monero_wallet(monero_network, &config, monero_daemon_host).await?;
+                monero_network,
+                &config,
+                monero_daemon_host,
+                execution_params,
+            )
+            .await?;
             let bitcoin_wallet = Arc::new(bitcoin_wallet);
             let (event_loop, mut event_loop_handle) = EventLoop::new(
                 &seed.derive_libp2p_identity(),
@@ -184,8 +189,13 @@ async fn main() -> Result<()> {
             }
 
             let bitcoin_wallet = init_bitcoin_wallet(bitcoin_network, &config, seed).await?;
-            let (monero_wallet, _process) =
+            let (monero_wallet, _process) = init_monero_wallet(
-                init_monero_wallet(monero_network, &config, monero_daemon_host).await?;
+                monero_network,
+                &config,
+                monero_daemon_host,
+                execution_params,
+            )
+            .await?;
             let bitcoin_wallet = Arc::new(bitcoin_wallet);
 
             let (event_loop, event_loop_handle) = EventLoop::new(
@@ -282,6 +292,7 @@ async fn init_monero_wallet(
     monero_network: monero::Network,
     config: &Config,
     monero_daemon_host: String,
+    execution_params: ExecutionParams,
 ) -> Result<(monero::Wallet, monero::WalletRpcProcess)> {
     const MONERO_BLOCKCHAIN_MONITORING_WALLET_NAME: &str = "swap-tool-blockchain-monitoring-wallet";
 
@@ -295,6 +306,7 @@ async fn init_monero_wallet(
         monero_wallet_rpc_process.endpoint(),
         monero_network,
         MONERO_BLOCKCHAIN_MONITORING_WALLET_NAME.to_string(),
+        execution_params.monero_avg_block_time,
     );
 
     monero_wallet.open_or_create().await?;

swap/src/execution_params.rs

@@ -1,12 +1,13 @@
 use crate::bitcoin::{CancelTimelock, PunishTimelock};
-use conquer_once::Lazy;
 use std::time::Duration;
+use time::NumericalStdDurationShort;
 
 #[derive(Debug, Copy, Clone)]
 pub struct ExecutionParams {
     pub bob_time_to_act: Duration,
     pub bitcoin_finality_confirmations: u32,
     pub bitcoin_avg_block_time: Duration,
+    pub monero_avg_block_time: Duration,
     pub monero_finality_confirmations: u32,
     pub bitcoin_cancel_timelock: CancelTimelock,
     pub bitcoin_punish_timelock: PunishTimelock,
@@ -28,12 +29,13 @@ pub struct Regtest;
 impl GetExecutionParams for Mainnet {
     fn get_execution_params() -> ExecutionParams {
         ExecutionParams {
-            bob_time_to_act: *mainnet::BOB_TIME_TO_ACT,
+            bob_time_to_act: 10.minutes(),
-            bitcoin_finality_confirmations: mainnet::BITCOIN_FINALITY_CONFIRMATIONS,
+            bitcoin_finality_confirmations: 3,
-            bitcoin_avg_block_time: *mainnet::BITCOIN_AVG_BLOCK_TIME,
+            bitcoin_avg_block_time: 10.minutes(),
-            monero_finality_confirmations: mainnet::MONERO_FINALITY_CONFIRMATIONS,
+            monero_avg_block_time: 2.minutes(),
-            bitcoin_cancel_timelock: mainnet::BITCOIN_CANCEL_TIMELOCK,
+            monero_finality_confirmations: 15,
-            bitcoin_punish_timelock: mainnet::BITCOIN_PUNISH_TIMELOCK,
+            bitcoin_cancel_timelock: CancelTimelock::new(72),
+            bitcoin_punish_timelock: PunishTimelock::new(72),
         }
     }
 }
@@ -41,12 +43,13 @@ impl GetExecutionParams for Mainnet {
 impl GetExecutionParams for Testnet {
     fn get_execution_params() -> ExecutionParams {
         ExecutionParams {
-            bob_time_to_act: *testnet::BOB_TIME_TO_ACT,
+            bob_time_to_act: 60.minutes(),
-            bitcoin_finality_confirmations: testnet::BITCOIN_FINALITY_CONFIRMATIONS,
+            bitcoin_finality_confirmations: 1,
-            bitcoin_avg_block_time: *testnet::BITCOIN_AVG_BLOCK_TIME,
+            bitcoin_avg_block_time: 5.minutes(),
-            monero_finality_confirmations: testnet::MONERO_FINALITY_CONFIRMATIONS,
+            monero_avg_block_time: 2.minutes(),
-            bitcoin_cancel_timelock: testnet::BITCOIN_CANCEL_TIMELOCK,
+            monero_finality_confirmations: 10,
-            bitcoin_punish_timelock: testnet::BITCOIN_PUNISH_TIMELOCK,
+            bitcoin_cancel_timelock: CancelTimelock::new(12),
+            bitcoin_punish_timelock: PunishTimelock::new(6),
         }
     }
 }
@@ -54,63 +57,13 @@ impl GetExecutionParams for Testnet {
 impl GetExecutionParams for Regtest {
     fn get_execution_params() -> ExecutionParams {
         ExecutionParams {
-            bob_time_to_act: *regtest::BOB_TIME_TO_ACT,
+            bob_time_to_act: 30.seconds(),
-            bitcoin_finality_confirmations: regtest::BITCOIN_FINALITY_CONFIRMATIONS,
+            bitcoin_finality_confirmations: 1,
-            bitcoin_avg_block_time: *regtest::BITCOIN_AVG_BLOCK_TIME,
+            bitcoin_avg_block_time: 5.seconds(),
-            monero_finality_confirmations: regtest::MONERO_FINALITY_CONFIRMATIONS,
+            monero_avg_block_time: 1.seconds(),
-            bitcoin_cancel_timelock: regtest::BITCOIN_CANCEL_TIMELOCK,
+            monero_finality_confirmations: 10,
-            bitcoin_punish_timelock: regtest::BITCOIN_PUNISH_TIMELOCK,
+            bitcoin_cancel_timelock: CancelTimelock::new(100),
+            bitcoin_punish_timelock: PunishTimelock::new(50),
         }
     }
 }
-
-mod mainnet {
-    use crate::execution_params::*;
-
-    // For each step, we are giving Bob 10 minutes to act.
-    pub static BOB_TIME_TO_ACT: Lazy<Duration> = Lazy::new(|| Duration::from_secs(10 * 60));
-
-    pub static BITCOIN_FINALITY_CONFIRMATIONS: u32 = 3;
-
-    pub static BITCOIN_AVG_BLOCK_TIME: Lazy<Duration> = Lazy::new(|| Duration::from_secs(10 * 60));
-
-    pub static MONERO_FINALITY_CONFIRMATIONS: u32 = 15;
-
-    // Set to 12 hours, arbitrary value to be reviewed properly
-    pub static BITCOIN_CANCEL_TIMELOCK: CancelTimelock = CancelTimelock::new(72);
-    pub static BITCOIN_PUNISH_TIMELOCK: PunishTimelock = PunishTimelock::new(72);
-}
-
-mod testnet {
-    use crate::execution_params::*;
-
-    pub static BOB_TIME_TO_ACT: Lazy<Duration> = Lazy::new(|| Duration::from_secs(60 * 60));
-
-    // This does not reflect recommended values for mainnet!
-    pub static BITCOIN_FINALITY_CONFIRMATIONS: u32 = 1;
-
-    pub static BITCOIN_AVG_BLOCK_TIME: Lazy<Duration> = Lazy::new(|| Duration::from_secs(5 * 60));
-
-    pub static MONERO_FINALITY_CONFIRMATIONS: u32 = 10;
-
-    // This does not reflect recommended values for mainnet!
-    pub static BITCOIN_CANCEL_TIMELOCK: CancelTimelock = CancelTimelock::new(12);
-    pub static BITCOIN_PUNISH_TIMELOCK: PunishTimelock = PunishTimelock::new(6);
-}
-
-mod regtest {
-    use crate::execution_params::*;
-
-    // In test, we set a shorter time to fail fast
-    pub static BOB_TIME_TO_ACT: Lazy<Duration> = Lazy::new(|| Duration::from_secs(30));
-
-    pub static BITCOIN_FINALITY_CONFIRMATIONS: u32 = 1;
-
-    pub static BITCOIN_AVG_BLOCK_TIME: Lazy<Duration> = Lazy::new(|| Duration::from_secs(5));
-
-    pub static MONERO_FINALITY_CONFIRMATIONS: u32 = 10;
-
-    pub static BITCOIN_CANCEL_TIMELOCK: CancelTimelock = CancelTimelock::new(100);
-
-    pub static BITCOIN_PUNISH_TIMELOCK: PunishTimelock = PunishTimelock::new(50);
-}

swap/src/monero/wallet.rs

@@ -3,15 +3,14 @@ use crate::monero::{
 };
 use ::monero::{Address, Network, PrivateKey, PublicKey};
 use anyhow::{Context, Result};
-use backoff::backoff::Constant as ConstantBackoff;
-use backoff::future::retry;
-use bitcoin::hashes::core::sync::atomic::AtomicU32;
 use monero_rpc::wallet;
-use monero_rpc::wallet::{BlockHeight, Refreshed};
+use monero_rpc::wallet::{BlockHeight, CheckTxKey, Refreshed};
+use std::cmp::min;
+use std::future::Future;
 use std::str::FromStr;
-use std::sync::atomic::Ordering;
 use std::time::Duration;
 use tokio::sync::Mutex;
+use tokio::time::Interval;
 use tracing::{debug, info};
 use url::Url;
 
@@ -20,22 +19,30 @@ pub struct Wallet {
     inner: Mutex<wallet::Client>,
     network: Network,
     name: String,
+    avg_block_time: Duration,
 }
 
 impl Wallet {
-    pub fn new(url: Url, network: Network, name: String) -> Self {
+    pub fn new(url: Url, network: Network, name: String, avg_block_time: Duration) -> Self {
         Self {
             inner: Mutex::new(wallet::Client::new(url)),
             network,
             name,
+            avg_block_time,
         }
     }
 
-    pub fn new_with_client(client: wallet::Client, network: Network, name: String) -> Self {
+    pub fn new_with_client(
+        client: wallet::Client,
+        network: Network,
+        name: String,
+        avg_block_time: Duration,
+    ) -> Self {
         Self {
             inner: Mutex::new(client),
             network,
             name,
+            avg_block_time,
         }
     }
 
@@ -157,63 +164,33 @@ impl Wallet {
         public_spend_key: PublicKey,
         public_view_key: PublicViewKey,
         transfer_proof: TransferProof,
-        expected_amount: Amount,
+        expected: Amount,
         conf_target: u32,
     ) -> Result<(), InsufficientFunds> {
-        let txid = &transfer_proof.tx_hash();
+        let txid = transfer_proof.tx_hash();
 
         tracing::info!(%txid, "Waiting for {} confirmation{} of Monero transaction", conf_target, if conf_target > 1 { "s" } else { "" });
 
-        enum Error {
-            TxNotFound,
-            InsufficientConfirmations,
-            InsufficientFunds { expected: Amount, actual: Amount },
-        }
-
         let address = Address::standard(self.network, public_spend_key, public_view_key.into());
 
-        let confirmations = AtomicU32::new(0u32);
+        let check_interval =
+            tokio::time::interval(min(self.avg_block_time / 10, Duration::from_secs(1)));
+        let key = &transfer_proof.tx_key().to_string();
 
-        let res = retry(ConstantBackoff::new(Duration::from_secs(1)), || async {
+        wait_for_confirmations(
-            // NOTE: Currently, this is conflicting IO errors with the transaction not being
+            txid.0,
-            // in the blockchain yet, or not having enough confirmations on it. All these
+            |txid| async move {
-            // errors warrant a retry, but the strategy should probably differ per case
+                self.inner
-            let proof = self
+                    .lock()
-                .inner
+                    .await
-                .lock()
+                    .check_tx_key(&txid, &key, &address.to_string())
-                .await
+                    .await
-                .check_tx_key(
+            },
-                    &String::from(transfer_proof.tx_hash()),
+            check_interval,
-                    &transfer_proof.tx_key().to_string(),
+            expected,
-                    &address.to_string(),
+            conf_target,
-                )
+        )
-                .await
+        .await?;
-                .map_err(|_| backoff::Error::Transient(Error::TxNotFound))?;
-
-            if proof.received != expected_amount.as_piconero() {
-                return Err(backoff::Error::Permanent(Error::InsufficientFunds {
-                    expected: expected_amount,
-                    actual: Amount::from_piconero(proof.received),
-                }));
-            }
-
-            if proof.confirmations >= confirmations.load(Ordering::SeqCst) {
-                confirmations.store(proof.confirmations, Ordering::SeqCst);
-
-                info!(%txid, "Monero lock tx has {} out of {} confirmations", proof.confirmations, conf_target);
-            }
-
-            if proof.confirmations < conf_target {
-                return Err(backoff::Error::Transient(Error::InsufficientConfirmations));
-            }
-
-            Ok(proof)
-        })
-        .await;
-
-        if let Err(Error::InsufficientFunds { expected, actual }) = res {
-            return Err(InsufficientFunds { expected, actual });
-        };
 
         Ok(())
     }
@@ -255,3 +232,124 @@ impl Wallet {
         Amount::from_monero(0.000_03f64).expect("static fee to be convertible without problems")
     }
 }
+
+async fn wait_for_confirmations<Fut>(
+    txid: String,
+    fetch_tx: impl Fn(String) -> Fut,
+    mut check_interval: Interval,
+    expected: Amount,
+    conf_target: u32,
+) -> Result<(), InsufficientFunds>
+where
+    Fut: Future<Output = Result<CheckTxKey>>,
+{
+    let mut seen_confirmations = 0u32;
+
+    while seen_confirmations < conf_target {
+        let tx = match fetch_tx(txid.clone()).await {
+            Ok(proof) => proof,
+            Err(error) => {
+                tracing::debug!(%txid, "Failed to retrieve tx from blockchain: {:#}", error);
+                continue; // treating every error as transient and retrying
+                          // is obviously wrong but the jsonrpc client is
+                          // too primitive to differentiate between all the
+                          // cases
+            }
+        };
+
+        let received = Amount::from_piconero(tx.received);
+
+        if received != expected {
+            return Err(InsufficientFunds {
+                expected,
+                actual: received,
+            });
+        }
+
+        if tx.confirmations > seen_confirmations {
+            seen_confirmations = tx.confirmations;
+            info!(%txid, "Monero lock tx has {} out of {} confirmations", tx.confirmations, conf_target);
+        }
+
+        check_interval.tick().await;
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use monero_rpc::wallet::CheckTxKey;
+    use std::sync::atomic::{AtomicU32, Ordering};
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn given_exact_confirmations_does_not_fetch_tx_again() {
+        let requests = Arc::new(AtomicU32::new(0));
+
+        let result = wait_for_confirmations(
+            String::from("TXID"),
+            move |_| {
+                let requests = requests.clone();
+
+                async move {
+                    match requests.fetch_add(1, Ordering::SeqCst) {
+                        0 => Ok(CheckTxKey {
+                            confirmations: 10,
+                            received: 100,
+                        }),
+                        _ => panic!("should not be called more than once"),
+                    }
+                }
+            },
+            tokio::time::interval(Duration::from_millis(10)),
+            Amount::from_piconero(100),
+            10,
+        )
+        .await;
+
+        assert!(result.is_ok())
+    }
+
+    /// A test that allows us to easily, visually verify if the log output is as
+    /// we desire.
+    ///
+    /// We want the following properties:
+    /// - Only print confirmations if they changed i.e. not every time we
+    ///   request them
+    /// - Also print the last one, i.e. 10 / 10
+    #[tokio::test]
+    async fn visual_log_check() {
+        let _ = tracing_subscriber::fmt().with_test_writer().try_init();
+        const MAX_REQUESTS: u32 = 20;
+
+        let requests = Arc::new(AtomicU32::new(0));
+
+        let result = wait_for_confirmations(
+            String::from("TXID"),
+            move |_| {
+                let requests = requests.clone();
+
+                async move {
+                    match requests.fetch_add(1, Ordering::SeqCst) {
+                        requests if requests <= MAX_REQUESTS => {
+                            Ok(CheckTxKey {
+                                confirmations: requests / 2, /* every 2nd request "yields" a
+                                                              * confirmation */
+                                received: 100,
+                            })
+                        }
+                        _ => panic!("should not be called more than {} times", MAX_REQUESTS),
+                    }
+                }
+            },
+            tokio::time::interval(Duration::from_millis(10)),
+            Amount::from_piconero(100),
+            10,
+        )
+        .await;
+
+        assert!(result.is_ok())
+    }
+}

swap/tests/testutils/mod.rs

@@ -359,6 +359,7 @@ where
         electrs_rpc_port,
         electrs_http_port,
         alice_seed,
+        execution_params,
     )
     .await;
 
@@ -381,6 +382,7 @@ where
         electrs_rpc_port,
         electrs_http_port,
         bob_seed,
+        execution_params,
     )
     .await;
 
@@ -590,6 +592,7 @@ async fn init_test_wallets(
     electrum_rpc_port: u16,
     electrum_http_port: u16,
     seed: Seed,
+    execution_params: ExecutionParams,
 ) -> (Arc<bitcoin::Wallet>, Arc<monero::Wallet>) {
     monero
         .init(vec![(name, starting_balances.xmr.as_piconero())])
@@ -600,6 +603,7 @@ async fn init_test_wallets(
         monero.wallet(name).unwrap().client(),
         monero::Network::default(),
         name.to_string(),
+        execution_params.monero_avg_block_time,
     );
 
     let electrum_rpc_url = {