Repository: farcaster-project/farcaster-node
Pull request #420 (2022-04-29 19:13:49)

Peer info: Use remote address to assemble internal id
 📗 📗 📗 📗🔴🔶


Patch:
diff --git a/Cargo.lock b/Cargo.lock
index b55a1cbd..3bffe539 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1772,7 +1772,7 @@ dependencies = [
 [[package]]
 name = "monero-rpc"
 version = "0.1.0"
-source = "git+https://github.com/monero-ecosystem/monero-rpc-rs?branch=master#29878ff67bba414cb0bbc334a7bbb15e7cd55e36"
+source = "git+https://github.com/monero-ecosystem/monero-rpc-rs?branch=master#dcda1161f279f345c95cde783f0d002ab2ae93cf"
 dependencies = [
  "anyhow",
  "chrono",
diff --git a/src/bin/peerd.rs b/src/bin/peerd.rs
index a6681662..7e00e0a5 100644
--- a/src/bin/peerd.rs
+++ b/src/bin/peerd.rs
@@ -229,7 +229,7 @@ fn main() {
 
                     internal_id = NodeAddr::Remote(RemoteNodeAddr {
                         node_id: opts.peer_key_opts.internal_node().node_id(),
-                        remote_addr: RemoteSocketAddr::Ftcp(inet_addr),
+                        remote_addr: RemoteSocketAddr::Ftcp(remote_socket),
                     });
                     debug!(
                         "Session successfully established with new unique id: {}",
@@ -257,6 +257,20 @@ fn main() {
     };
 
     debug!("Starting runtime ...");
+
+    /* A maker / listener passes the following content
+        internal_id: local key and remote address
+        remote_id: None
+        local_socket: local inet address
+        remote_socket: address of the remote socket
+        connect: false
+
+    A taker / connecter passes the following content
+        internal_id: remote key and remote address
+        remote_id: remote peer id
+        local_socket: None
+        remote_socket: remote node addr
+        connect: true */
     peerd::run(
         service_config,
         connection,
diff --git a/src/config.rs b/src/config.rs
index 0eb3f086..5573e22f 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -150,8 +150,6 @@ pub struct SyncerServers {
     pub monero_rpc_wallet: String,
     /// Monero lws to use
     pub monero_lws: Option<String>,
-    /// Monero wallet directory
-    pub monero_wallet_dir: Option<String>,
 }
 
 impl Default for SyncersConfig {
@@ -162,14 +160,12 @@ impl Default for SyncersConfig {
                 monero_daemon: FARCASTER_MAINNET_MONERO_DAEMON.into(),
                 monero_rpc_wallet: FARCASTER_MAINNET_MONERO_RPC_WALLET.into(),
                 monero_lws: None,
-                monero_wallet_dir: None,
             }),
             testnet: Some(SyncerServers {
                 electrum_server: FARCASTER_TESTNET_ELECTRUM_SERVER.into(),
                 monero_daemon: FARCASTER_TESTNET_MONERO_DAEMON.into(),
                 monero_rpc_wallet: FARCASTER_TESTNET_MONERO_RPC_WALLET.into(),
                 monero_lws: None,
-                monero_wallet_dir: None,
             }),
             local: None,
         }
diff --git a/src/farcasterd/runtime.rs b/src/farcasterd/runtime.rs
index 69ec61de..35ef6dd1 100644
--- a/src/farcasterd/runtime.rs
+++ b/src/farcasterd/runtime.rs
@@ -1166,9 +1166,11 @@ impl Runtime {
                                 monero_rpc::RpcClient::new(host);
                             let wallet = wallet_client.wallet();
                             let options = monero_rpc::TransferOptions::default();
+                            let mut destination = HashMap::new();
+                            destination.insert(address, amount.as_pico());
                             match wallet
                                 .transfer(
-                                    [(address, amount)].iter().cloned().collect(),
+                                    destination.clone(),
                                     monero_rpc::TransferPriority::Default,
                                     options.clone(),
                                 )
@@ -1284,6 +1286,17 @@ impl Runtime {
                 )?;
             }
 
+            Request::PeerdTerminated => {
+                if let ServiceId::Peer(addr) = source {
+                    if self.connections.remove(&addr) {
+                        debug!(
+                            "removed connection {} from farcasterd registered connections",
+                            addr
+                        );
+                    }
+                }
+            }
+
             req => {
                 error!("Ignoring unsupported request: {}", req.err());
             }
@@ -1507,11 +1520,6 @@ fn syncer_servers_args(config: &Config, coin: Coin, net: Network) -> Result<Vec<
                         .monero_lws
                         .map_or(vec![], |v| vec!["--monero-lws".to_string(), v]),
                 );
-                args.extend(
-                    servers
-                        .monero_wallet_dir
-                        .map_or(vec![], |v| vec!["--monero-wallet-dir-path".to_string(), v]),
-                );
                 Ok(args)
             }
         },
diff --git a/src/peerd/runtime.rs b/src/peerd/runtime.rs
index e2c41d38..76fc712b 100644
--- a/src/peerd/runtime.rs
+++ b/src/peerd/runtime.rs
@@ -229,8 +229,9 @@ impl esb::Handler<ServiceBus> for Runtime {
     ) -> Result<(), Error> {
         if self.connect {
             info!(
-                "{} with the remote peer",
-                "Initializing connection".bright_blue_bold()
+                "{} with the remote peer {}",
+                "Initializing connection".bright_blue_bold(),
+                self.remote_socket
             );
             // self.send_ctl(senders, ServiceId::Wallet, request::PeerSecret)
 
@@ -392,7 +393,12 @@ impl Runtime {
 
             Request::Protocol(Msg::PeerdShutdown) => {
                 warn!("Exiting peerd");
-                // handle further side effects here, e.g. sending a message to farcasterd to cleanup the peer connection, before exiting.
+                senders.send_to(
+                    ServiceBus::Ctl,
+                    self.identity(),
+                    ServiceId::Farcasterd,
+                    Request::PeerdTerminated,
+                )?;
                 std::process::exit(0);
             }
 
diff --git a/src/rpc/request.rs b/src/rpc/request.rs
index 82f9de0c..5e6979b2 100644
--- a/src/rpc/request.rs
+++ b/src/rpc/request.rs
@@ -347,6 +347,10 @@ pub enum Request {
     #[display("terminate")]
     Terminate,
 
+    #[api(type = 4)]
+    #[display("peerd_terminated")]
+    PeerdTerminated,
+
     #[api(type = 32)]
     #[display("nodeid({0})")]
     NodeId(NodeId),
diff --git a/src/syncerd/monero_syncer.rs b/src/syncerd/monero_syncer.rs
index d4774e8f..eac6dd89 100644
--- a/src/syncerd/monero_syncer.rs
+++ b/src/syncerd/monero_syncer.rs
@@ -28,9 +28,7 @@ use monero_rpc::{
     PrivateKeyType, TransferType,
 };
 use std::collections::HashMap;
-use std::fs;
 use std::ops::Range;
-use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::mpsc::{Receiver, TryRecvError};
 use std::sync::Arc;
@@ -258,13 +256,18 @@ async fn sweep_address(
     network: &monero::Network,
     wallet_mutex: Arc<Mutex<monero_rpc::WalletClient>>,
     restore_height: Option<u64>,
-    wallet_dir_path: Option<PathBuf>,
 ) -> Result<Vec<Vec<u8>>, Error> {
     let keypair = monero::KeyPair { view, spend };
     let password = s!(" ");
     let address = monero::Address::from_keypair(*network, &keypair);
     let wallet_filename = format!("sweep:{}", address);
 
+    // add some extra leeway for the wallet restore height
+    // let restore_height = match from_height {
+    //     Some(height) if height > 10 => Some(height - 10),
+    //     _ => None,
+    // };
+
     let wallet = wallet_mutex.lock().await;
     trace!("taking sweep wallet lock");
 
@@ -330,42 +333,6 @@ async fn sweep_address(
             })
             .collect();
 
-        // close the wallet since we are done with it now
-        wallet.close_wallet().await?;
-
-        // delete all the associated wallet data if possible
-        if let Some(path) = wallet_dir_path {
-            if let Some(raw_path) = path.to_str() {
-                let watch_wallet_filename = format!("/watch:{}", address);
-                let sweep_wallet_filename = format!("/sweep:{}", address);
-                if let Err(error) = fs::remove_file(raw_path.to_string() + &sweep_wallet_filename)
-                    .and(fs::remove_file(
-                        raw_path.to_string() + &sweep_wallet_filename + &".keys".to_string(),
-                    ))
-                {
-                    warn!("Failed to clean sweep wallet data after successful sweep. {}. The path used for the wallet directory is probably malformed", error);
-                } else {
-                    info!("Successfully removed sweep wallet data after completed sweep.");
-                }
-
-                if let Err(error) = fs::remove_file(raw_path.to_string() + &watch_wallet_filename)
-                    .and(fs::remove_file(
-                        raw_path.to_string() + &watch_wallet_filename + &".address.txt".to_string(),
-                    ))
-                    .and(fs::remove_file(
-                        raw_path.to_string() + &watch_wallet_filename + &".keys".to_string(),
-                    ))
-                {
-                    debug!("Failed to clean watch-only wallet data after sweep. {}. The path used for the wallet directory is probably malformed", error);
-                } else {
-                    debug!("Successfully removed watch-only wallet data after completed sweep");
-                }
-            } else {
-                warn!("No associated wallet data cleaned up after sweep. The path used for the wallet directory is probably malformed");
-            }
-        } else {
-            info!("{}", format!("Completed operations on Monero wallets with address {} . These wallets can now be safely deleted", address));
-        }
         Ok(tx_ids)
     } else {
         debug!(
@@ -652,7 +619,6 @@ fn sweep_polling(
     state: Arc<Mutex<SyncerState>>,
     wallet: Arc<Mutex<monero_rpc::WalletClient>>,
     network: monero::Network,
-    wallet_dir_path: Option<PathBuf>,
 ) -> tokio::task::JoinHandle<()> {
     tokio::task::spawn(async move {
         loop {
@@ -669,7 +635,6 @@ fn sweep_polling(
                         &network,
                         Arc::clone(&wallet),
                         sweep_address_task.from_height,
-                        wallet_dir_path.clone(),
                     )
                     .await
                     .unwrap_or_else(|err| {
@@ -789,10 +754,6 @@ impl Synclet for MoneroSyncer {
                     monero_lws: opts.monero_lws.clone(),
                 };
                 info!("monero syncer servers: {:?}", syncer_servers);
-                let wallet_dir = opts
-                    .monero_wallet_dir_path
-                    .clone()
-                    .map(|wallet_dir| PathBuf::from(wallet_dir));
 
                 let _handle = std::thread::spawn(move || {
                     use tokio::runtime::Builder;
@@ -835,12 +796,8 @@ impl Synclet for MoneroSyncer {
                         let unseen_transaction_handle =
                             unseen_transaction_polling(Arc::clone(&state), syncer_servers.clone());
 
-                        let sweep_handle = sweep_polling(
-                            Arc::clone(&state),
-                            Arc::clone(&wallet_mutex),
-                            network,
-                            wallet_dir,
-                        );
+                        let sweep_handle =
+                            sweep_polling(Arc::clone(&state), Arc::clone(&wallet_mutex), network);
 
                         let res = tokio::try_join!(
                             address_handle,
diff --git a/src/syncerd/opts.rs b/src/syncerd/opts.rs
index 419a0441..250bcf8c 100644
--- a/src/syncerd/opts.rs
+++ b/src/syncerd/opts.rs
@@ -65,10 +65,6 @@ pub struct Opts {
     /// Monero lws to use for Monero syncers
     #[clap(long)]
     pub monero_lws: Option<String>,
-
-    /// Wallet directory use by the monero-wallet-rpc
-    #[clap(long)]
-    pub monero_wallet_dir_path: Option<String>,
 }
 
 #[derive(Clap, Display, Copy, Clone, Hash, PartialEq, Eq, Debug, StrictEncode, StrictDecode)]


Test patch:
diff --git a/tests/functional-swap.rs b/tests/functional-swap.rs
index d788befe..f02ae61a 100644
--- a/tests/functional-swap.rs
+++ b/tests/functional-swap.rs
@@ -1959,6 +1959,8 @@ async fn send_monero(
     address: monero::Address,
     amount: monero::Amount,
 ) -> Vec<u8> {
+    let mut destination: HashMap<monero::Address, u64> = HashMap::new();
+    destination.insert(address, amount.as_pico());
     let options = monero_rpc::TransferOptions {
         account_index: None,
         subaddr_indices: None,
@@ -1976,7 +1978,7 @@ async fn send_monero(
     wallet_lock.refresh(Some(1)).await.unwrap();
     let transaction = wallet_lock
         .transfer(
-            [(address, amount)].iter().cloned().collect(),
+            destination.clone(),
             monero_rpc::TransferPriority::Default,
             options.clone(),
         )
diff --git a/tests/functional.rs b/tests/functional.rs
index fbd5e4f5..3b91496c 100644
--- a/tests/functional.rs
+++ b/tests/functional.rs
@@ -16,9 +16,11 @@ use internet2::Decrypt;
 use internet2::PlainTranscoder;
 use internet2::RoutedFrame;
 use internet2::ZMQ_CONTEXT;
+use monero::Address;
 use monero_rpc::GetBlockHeaderSelector;
 use paste::paste;
 use rand::{distributions::Alphanumeric, Rng};
+use std::collections::HashMap;
 use std::sync::mpsc::Receiver;
 use std::sync::mpsc::Sender;
 
@@ -1472,6 +1474,8 @@ async fn monero_syncer_transaction_test() {
     let request = get_request_from_message(message);
     assert_transaction_confirmations(request, Some(0), vec![0]);
 
+    let mut destination: HashMap<Address, u64> = HashMap::new();
+    destination.insert(address, 1000);
     let options = monero_rpc::TransferOptions {
         account_index: None,
         subaddr_indices: None,
@@ -1483,10 +1487,7 @@ async fn monero_syncer_transaction_test() {
     };
     let transaction = wallet
         .transfer(
-            [(address, monero::Amount::from_pico(1000))]
-                .iter()
-                .cloned()
-                .collect(),
+            destination.clone(),
             monero_rpc::TransferPriority::Default,
             options.clone(),
         )
@@ -1782,6 +1783,8 @@ async fn send_monero(
     address: monero::Address,
     amount: u64,
 ) -> Vec<u8> {
+    let mut destination: HashMap<Address, u64> = HashMap::new();
+    destination.insert(address, amount);
     let options = monero_rpc::TransferOptions {
         account_index: None,
         subaddr_indices: None,
@@ -1793,10 +1796,7 @@ async fn send_monero(
     };
     let transaction = wallet
         .transfer(
-            [(address, monero::Amount::from_pico(amount))]
-                .iter()
-                .cloned()
-                .collect(),
+            destination.clone(),
             monero_rpc::TransferPriority::Default,
             options.clone(),
         )