Repository: farcaster-project/farcaster-node
Pull request #154 (2021-10-28 15:01:46)

Monero sweep0


Patch:
diff --git a/Cargo.lock b/Cargo.lock
index ca189112..d971da7f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1732,7 +1732,7 @@ dependencies = [
 [[package]]
 name = "monero-rpc"
 version = "0.1.0"
-source = "git+https://github.com/TheCharlatan/monero-rpc-rs?branch=changes#5b311c820d3420f66a4aa7b546662be65c6df316"
+source = "git+https://github.com/TheCharlatan/monero-rpc-rs?branch=changes#467cc0c66b906abfc4c4f5ecb88eb477c75f2f9e"
 dependencies = [
  "anyhow",
  "chrono",
diff --git a/src/cli/command.rs b/src/cli/command.rs
index 4f0557af..34370551 100644
--- a/src/cli/command.rs
+++ b/src/cli/command.rs
@@ -15,6 +15,7 @@
 use std::{
     convert::TryFrom,
     io::{self, Read, Write},
+    sync::mpsc,
     time::Duration,
 };
 use std::{str::FromStr, thread::sleep};
@@ -204,7 +205,7 @@ impl Exec for Command {
             }
             Command::Progress { swapid } => {
                 runtime.request(ServiceId::Farcasterd, Request::ReadProgress(swapid))?;
-                while let Ok(n) = runtime.report_progress() {}
+                runtime.report_progress()?;
             }
             _ => unimplemented!(),
         }
@@ -225,4 +226,3 @@ fn take_offer() -> bool {
         _ => take_offer(),
     }
 }
-
diff --git a/src/cli/opts.rs b/src/cli/opts.rs
index f32f5737..92d40913 100644
--- a/src/cli/opts.rs
+++ b/src/cli/opts.rs
@@ -174,6 +174,7 @@ pub enum Command {
         #[clap(short, long)]
         without_validation: bool,
     },
+
     /// Request Swap progress report
     Progress {
         /// SwapId
diff --git a/src/farcasterd/runtime.rs b/src/farcasterd/runtime.rs
index 585fe798..dfb787a3 100644
--- a/src/farcasterd/runtime.rs
+++ b/src/farcasterd/runtime.rs
@@ -213,7 +213,8 @@ impl Runtime {
                         &public_offer
                     );
                     if let Some(arb_addr) = self.arb_addrs.remove(&public_offer.id()) {
-                        let btc_addr_req = Request::BitcoinAddress(BitcoinAddress(*swap_id, arb_addr));
+                        let btc_addr_req =
+                            Request::BitcoinAddress(BitcoinAddress(*swap_id, arb_addr));
                         senders.send_to(
                             ServiceBus::Msg,
                             self.identity(),
@@ -680,12 +681,12 @@ impl Runtime {
                     let PublicOffer {
                         version,
                         offer,
-                        node_id, // bitcoin::Pubkey
-                        peer_address,   // InetSocketAddr
+                        node_id,      // bitcoin::Pubkey
+                        peer_address, // InetSocketAddr
                     } = public_offer.clone();
 
                     let daemon_service = internet2::RemoteNodeAddr {
-                        node_id,                           // checked above
+                        node_id,                                           // checked above
                         remote_addr: RemoteSocketAddr::Ftcp(peer_address), /* expected RemoteSocketAddr */
                     };
                     let peer = daemon_service
@@ -759,14 +760,30 @@ impl Runtime {
                 if !self.progress.contains_key(&source) {
                     self.progress.insert(source.clone(), none!());
                 };
+                let identity = self.identity();
                 let queue = self.progress.get_mut(&source).expect("checked/added above");
                 queue.push_back(request);
             }
             Request::ReadProgress(swapid) => {
                 let id = &ServiceId::Swap(swapid);
+                let identify = self.identity();
                 if let Some(queue) = self.progress.get_mut(id) {
-                    for req in queue.iter() {
-                        report_to.push((Some(source.clone()), req.clone()))
+                    let n = queue.len();
+
+                    for (i, req) in queue.iter().enumerate() {
+                        let x = match req {
+                            Request::Progress(x)
+                            | Request::Success(OptionDetails(Some(x)))
+                            | Request::Failure(Failure { code: _, info: x }) => x,
+                            _ => unreachable!("not handled here"),
+                        };
+                        let req = if i < n - 1 {
+                            Request::Progress(x.clone())
+                        } else {
+                            Request::Success(OptionDetails(Some(x.clone())))
+                        };
+                        report_to.push((Some(source.clone()), req));
+                        // senders.send_to(ServiceBus::Ctl, identify.clone(), source.clone(), req)?;
                     }
                 }
             }
diff --git a/src/rpc/request.rs b/src/rpc/request.rs
index 0d590239..2f6ccf91 100644
--- a/src/rpc/request.rs
+++ b/src/rpc/request.rs
@@ -12,7 +12,7 @@
 // along with this software.
 // If not, see <https://opensource.org/licenses/MIT>.
 
-use crate::syncerd::types::{Event, Task};
+use crate::syncerd::{SweepXmrAddress, types::{Event, Task}};
 use crate::walletd::NodeSecrets;
 use amplify::{Holder, ToYamlString, Wrapper};
 use farcaster_core::syncer::BroadcastTransaction;
@@ -488,19 +488,24 @@ pub enum Request {
     // #[from]
     // CreateTask(u64), // FIXME
     #[api(type = 1300)]
-    #[display("task({0})", alt = "{0:#}")]
+    #[display("syncer_task({0})", alt = "{0:#}")]
     #[from]
     SyncerTask(Task),
 
     #[api(type = 1301)]
-    #[display("task({0})", alt = "{0:#}")]
+    #[display("syncer_event({0})", alt = "{0:#}")]
     #[from]
     SyncerEvent(Event),
 
     #[api(type = 1302)]
-    #[display("task({0})", alt = "{0:#}")]
+    #[display("syncer_bridge_ev({0})", alt = "{0:#}")]
     #[from]
     SyncerdBridgeEvent(SyncerdBridgeEvent),
+
+    #[api(type = 1303)]
+    #[display("task({0})", alt = "{0:#}")]
+    #[from]
+    SweepXmrAddress(SweepXmrAddress),
 }
 
 impl rpc_connection::Request for Request {}
diff --git a/src/swapd/runtime.rs b/src/swapd/runtime.rs
index d1b1f4e2..58ced53d 100644
--- a/src/swapd/runtime.rs
+++ b/src/swapd/runtime.rs
@@ -13,7 +13,12 @@
 // along with this software.
 // If not, see <https://opensource.org/licenses/MIT>.
 
-use crate::syncerd::{opts::Coin, Abort, HeightChanged, WatchHeight, XmrAddressAddendum};
+use crate::{
+    syncerd::{
+        opts::Coin, Abort, HeightChanged, SweepAddress, SweepAddressAddendum, SweepXmrAddress,
+        WatchHeight, XmrAddressAddendum,
+    },
+};
 use std::{
     any::Any,
     collections::{BTreeMap, HashMap, HashSet},
@@ -75,7 +80,7 @@ use internet2::{
 };
 use lnpbp::chain::Chain;
 use microservices::esb::{self, Handler};
-use monero::cryptonote::hash::keccak_256;
+use monero::{cryptonote::hash::keccak_256, ViewPair};
 use request::{Commit, InitSwap, Params, Reveal, TakeCommit, Tx};
 
 pub fn run(
@@ -120,12 +125,6 @@ pub fn run(
         watched_addrs: none!(),
         watched_txs: none!(),
     };
-
-    let net = match network {
-        blockchain::Network::Mainnet => monero::Network::Mainnet,
-        blockchain::Network::Testnet => monero::Network::Stagenet,
-        blockchain::Network::Local => monero::Network::Stagenet,
-    };
     let syncer_state = SyncerState {
         tasks,
         monero_height: 0,
@@ -133,7 +132,7 @@ pub fn run(
         confirmation_bound: 50000,
         lock_tx_confs: None,
         cancel_tx_confs: None,
-        monero_address: Box::new(move |view_pair| monero::Address::from_viewpair(net, view_pair)),
+        network,
         bitcoin_syncer: ServiceId::Syncer(Coin::Bitcoin, network),
         monero_syncer: ServiceId::Syncer(Coin::Monero, network),
         monero_amount,
@@ -260,7 +259,7 @@ struct SyncerState {
     confirmation_bound: u32,
     lock_tx_confs: Option<Request>,
     cancel_tx_confs: Option<Request>,
-    monero_address: Box<dyn Fn(&monero::ViewPair) -> monero::Address>,
+    network: farcaster_core::blockchain::Network,
     bitcoin_syncer: ServiceId,
     monero_syncer: ServiceId,
     monero_amount: monero::Amount,
@@ -397,6 +396,7 @@ impl SyncerState {
             u64::MAX
         }
     }
+
     fn from_height(&self, coin: Coin) -> u64 {
         let lookback = match coin {
             Coin::Monero => 300,
@@ -488,26 +488,10 @@ impl SyncerState {
     fn watch_addr_xmr(
         &mut self,
         spend: monero::PublicKey,
-        accordant_shared_keys: Vec<TaggedElement<SharedKeyId, monero::PrivateKey>>,
+        view: monero::PrivateKey,
         swap_role: SwapRole,
         tx_label: TxLabel,
     ) -> Task {
-        let view = accordant_shared_keys
-            .into_iter()
-            .find(|vk| vk.tag() == &SharedKeyId::new(SHARED_VIEW_KEY_ID))
-            .unwrap()
-            .elem()
-            .clone();
-        if swap_role == SwapRole::Alice {
-            let viewpair = monero::ViewPair { spend, view };
-            let address = (self.monero_address)(&viewpair);
-            info!(
-                "Alice, please send {} to {}",
-                self.monero_amount.to_string().bright_green_bold(),
-                address.addr(),
-            );
-        }
-
         let from_height = self.from_height(Coin::Monero);
 
         let addendum = AddressAddendum::Monero(XmrAddressAddendum {
@@ -528,6 +512,27 @@ impl SyncerState {
         };
         Task::WatchAddress(watch_addr)
     }
+    fn sweep_xmr(
+        &mut self,
+        view_key: monero::PrivateKey,
+        spend_key: monero::PrivateKey,
+        address: String,
+    ) -> Task {
+        let id = self.tasks.new_taskid();
+        let lifetime = self.task_lifetime(Coin::Monero);
+        let addendum = SweepAddressAddendum::Monero(SweepXmrAddress {
+            view_key,
+            spend_key,
+            address,
+        });
+        let sweep_task = SweepAddress {
+            id,
+            lifetime,
+            addendum,
+        };
+        Task::SweepAddress(sweep_task)
+    }
+
     fn acc_lock_watched(&self) -> bool {
         self.tasks
             .watched_txs
@@ -591,7 +596,8 @@ impl Runtime {
             next_state.bright_white_bold()
         );
         self.state = next_state;
-        self.report_success_to(senders, self.enquirer.clone(), Some(msg))
+        self.report_success_to(senders, self.enquirer.clone(), Some(msg))?;
+        Ok(())
     }
 
     fn broadcast(
@@ -1053,6 +1059,20 @@ impl Runtime {
 
         let ctl_bus = ServiceBus::Ctl;
         match request {
+            Request::SweepXmrAddress(SweepXmrAddress {
+                view_key,
+                spend_key,
+                address,
+            }) if source == ServiceId::Wallet => {
+                let task = self.syncer_state.sweep_xmr(view_key, spend_key, address);
+
+                senders.send_to(
+                    ServiceBus::Ctl,
+                    self.identity(),
+                    self.syncer_state.monero_syncer(),
+                    Request::SyncerTask(task),
+                )?
+            }
             Request::TakeSwap(InitSwap {
                 peerd,
                 report_to,
@@ -1493,9 +1513,36 @@ impl Runtime {
                                             ..
                                         })) = self.remote_params.clone()
                                         {
+                                            let view = accordant_shared_keys
+                                                .into_iter()
+                                                .find(|vk| {
+                                                    vk.tag()
+                                                        == &SharedKeyId::new(SHARED_VIEW_KEY_ID)
+                                                })
+                                                .unwrap()
+                                                .elem()
+                                                .clone();
+                                            let viewpair = monero::ViewPair { spend, view };
+                                            let address = monero::Address::from_viewpair(
+                                                self.syncer_state.network.into(),
+                                                &viewpair,
+                                            );
+                                            info!(
+                                                "Alice, please send {} to {}",
+                                                self.syncer_state
+                                                    .monero_amount
+                                                    .to_string()
+                                                    .bright_green_bold(),
+                                                address.addr(),
+                                            );
+                                            self.report_success_to(
+                                                senders,
+                                                self.enquirer.clone(),
+                                                Some(address.to_string()),
+                                            )?;
                                             let watch_addr_task = self.syncer_state.watch_addr_xmr(
                                                 spend,
-                                                accordant_shared_keys,
+                                                view,
                                                 self.state.swap_role(),
                                                 TxLabel::AccLock,
                                             );
@@ -1608,6 +1655,9 @@ impl Runtime {
                     Event::TaskAborted(event) => {
                         info!("{}", event)
                     }
+                    Event::SweepSuccess(event) => {
+                        debug!("{}", event)
+                    }
                 }
             }
             Request::Protocol(Msg::CoreArbitratingSetup(core_arb_setup)) => {
@@ -1654,9 +1704,15 @@ impl Runtime {
                         ..
                     })) = self.remote_params.clone()
                     {
+                        let view = accordant_shared_keys
+                            .into_iter()
+                            .find(|vk| vk.tag() == &SharedKeyId::new(SHARED_VIEW_KEY_ID))
+                            .unwrap()
+                            .elem()
+                            .clone();
                         let task = self.syncer_state.watch_addr_xmr(
                             spend,
-                            accordant_shared_keys,
+                            view,
                             self.state.swap_role(),
                             TxLabel::AccLock,
                         );
diff --git a/src/syncerd/bitcoin_syncer.rs b/src/syncerd/bitcoin_syncer.rs
index fb2b77ea..9c33b483 100644
--- a/src/syncerd/bitcoin_syncer.rs
+++ b/src/syncerd/bitcoin_syncer.rs
@@ -338,6 +338,9 @@ async fn run_syncerd_task_receiver(
             match syncerd_task {
                 Ok(syncerd_task) => {
                     match syncerd_task.task {
+                        Task::SweepAddress(_) => {
+                            error!("sweep address not implemented for bitcoin syncer");
+                        }
                         Task::Abort(task) => {
                             let mut state_guard = state.lock().await;
                             state_guard.abort(task.id, syncerd_task.source).await;
diff --git a/src/syncerd/monero_syncer.rs b/src/syncerd/monero_syncer.rs
index a198cd04..d36a2ede 100644
--- a/src/syncerd/monero_syncer.rs
+++ b/src/syncerd/monero_syncer.rs
@@ -9,7 +9,7 @@ use crate::syncerd::runtime::SyncerdTask;
 use crate::syncerd::syncer_state::create_set;
 use crate::syncerd::syncer_state::AddressTx;
 use crate::syncerd::syncer_state::SyncerState;
-use crate::syncerd::types::{AddressAddendum, Task};
+use crate::syncerd::types::{AddressAddendum, SweepAddressAddendum, Task};
 use crate::syncerd::Event;
 use crate::syncerd::SyncerServers;
 use crate::syncerd::TransactionBroadcasted;
@@ -18,15 +18,14 @@ use internet2::zmqsocket::{Connection, ZmqType};
 use internet2::PlainTranscoder;
 use lnpbp::chain::Chain;
 use monero::Hash;
-use monero_rpc::GenerateFromKeysArgs;
-use monero_rpc::GetBlockHeaderSelector;
-use monero_rpc::GetTransfersCategory;
-use monero_rpc::GetTransfersSelector;
-use monero_rpc::TransferType;
+use monero_rpc::{
+    GenerateFromKeysArgs, GetBlockHeaderSelector, GetTransfersCategory, GetTransfersSelector,
+    TransferType,
+};
 use std::collections::HashMap;
 use std::ops::Range;
-use std::sync::mpsc::Receiver;
-use std::sync::mpsc::TryRecvError;
+use std::str::FromStr;
+use std::sync::mpsc::{Receiver, TryRecvError};
 use std::sync::Arc;
 use tokio::runtime::Runtime;
 use tokio::sync::mpsc::Receiver as TokioReceiver;
@@ -43,6 +42,7 @@ pub struct MoneroRpc {
     block_hash: Vec<u8>,
 }
 
+#[derive(Debug)]
 pub struct Block {
     height: u64,
     block_hash: Vec<u8>,
@@ -152,15 +152,17 @@ impl MoneroRpc {
         &mut self,
         address_addendum: XmrAddressAddendum,
         network: monero::Network,
+        wallet_mutex: Arc<Mutex<monero_rpc::WalletClient>>,
     ) -> Result<AddressNotif, Error> {
         let keypair = monero::ViewPair {
             spend: address_addendum.spend_key,
             view: address_addendum.view_key,
         };
         let address = monero::Address::from_viewpair(network, &keypair);
-        let wallet = monero_rpc::RpcClient::new(self.wallet_rpc_url.clone()).wallet();
         let password = s!(" ");
 
+        let wallet = wallet_mutex.lock().await;
+
         match wallet
             .open_wallet(address.to_string(), Some(password.clone()))
             .await
@@ -203,6 +205,7 @@ impl MoneroRpc {
         };
 
         let mut transfers = wallet.get_transfers(selector).await?;
+        drop(wallet);
 
         let mut address_txs: Vec<AddressTx> = vec![];
         for (_category, mut txs) in transfers.drain() {
@@ -220,6 +223,77 @@ impl MoneroRpc {
     }
 }
 
+async fn sweep_address(
+    dest_address: monero::Address,
+    view: monero::PrivateKey,
+    spend: monero::PrivateKey,
+    network: &monero::Network,
+    wallet_mutex: Arc<Mutex<monero_rpc::WalletClient>>,
+) -> Result<Option<Vec<Vec<u8>>>, Error> {
+    let keypair = monero::KeyPair { view, spend };
+    let password = s!(" ");
+    let address = monero::Address::from_keypair(*network, &keypair);
+
+    let wallet = wallet_mutex.lock().await;
+
+    match wallet
+        .open_wallet(address.to_string(), Some(password.clone()))
+        .await
+    {
+        Ok(_) => {}
+        Err(err) => {
+            debug!(
+                "error opening to be sweeped wallet: {:?}, falling back to generating a new wallet",
+                err,
+            );
+            wallet
+                .generate_from_keys(GenerateFromKeysArgs {
+                    restore_height: Some(1),
+                    filename: address.to_string(),
+                    address,
+                    spendkey: Some(keypair.spend),
+                    viewkey: keypair.view,
+                    password: password.clone(),
+                    autosave_current: Some(true),
+                })
+                .await?;
+            wallet
+                .open_wallet(address.to_string(), Some(password))
+                .await?;
+        }
+    }
+
+    wallet.refresh(Some(1)).await?;
+
+    let balance = wallet.get_balance(0, None).await?;
+    // only sweep once all the balance is unlocked
+    if balance.balance >= balance.unlocked_balance {
+        let sweep_args = monero_rpc::SweepAllArgs {
+            address: dest_address,
+            account_index: 0,
+            subaddr_indices: None,
+            priority: monero_rpc::TransferPriority::Default,
+            mixin: 10,
+            ring_size: 11,
+            unlock_time: 0,
+            get_tx_keys: None,
+            below_amount: None,
+            do_not_relay: None,
+            get_tx_hex: None,
+            get_tx_metadata: None,
+        };
+        let res = wallet.sweep_all(sweep_args).await?;
+        let tx_ids: Vec<Vec<u8>> = res
+            .tx_hash_list
+            .iter()
+            .map(|hash| hex::decode(hash.to_string()).unwrap())
+            .collect();
+
+        return Ok(Some(tx_ids));
+    }
+    Ok(None)
+}
+
 pub struct MoneroSyncer {}
 
 impl MoneroSyncer {
@@ -243,6 +317,18 @@ async fn run_syncerd_task_receiver(
             match syncerd_task {
                 Ok(syncerd_task) => {
                     match syncerd_task.task {
+                        Task::SweepAddress(task) => match task.addendum.clone() {
+                            SweepAddressAddendum::Monero(_) => {
+                                info!("sweeping to address: {}", task);
+                                let mut state_guard = state.lock().await;
+                                state_guard.sweep_address(task, syncerd_task.source);
+                            }
+                            _ => {
+                                error!("Aborting sweep address task - unable to decode sweep address addendum");
+                                let mut state_guard = state.lock().await;
+                                state_guard.abort(task.id, syncerd_task.source).await;
+                            }
+                        },
                         Task::Abort(task) => {
                             let mut state_guard = state.lock().await;
                             state_guard.abort(task.id, syncerd_task.source).await;
@@ -295,6 +381,7 @@ fn address_polling(
     state: Arc<Mutex<SyncerState>>,
     syncer_servers: SyncerServers,
     network: monero::Network,
+    wallet_mutex: Arc<Mutex<monero_rpc::WalletClient>>,
 ) -> tokio::task::JoinHandle<()> {
     tokio::task::spawn(async move {
         let mut rpc = MoneroRpc::new(
@@ -314,7 +401,11 @@ fn address_polling(
                 // wallet
                 let mut address_transactions = None;
                 match rpc
-                    .check_address(address_addendum.clone(), network.clone())
+                    .check_address(
+                        address_addendum.clone(),
+                        network.clone(),
+                        Arc::clone(&wallet_mutex),
+                    )
                     .await
                 {
                     Ok(addr_txs) => {
@@ -391,6 +482,48 @@ fn height_polling(
     })
 }
 
+fn sweep_polling(
+    state: Arc<Mutex<SyncerState>>,
+    wallet: Arc<Mutex<monero_rpc::WalletClient>>,
+    network: monero::Network,
+) -> tokio::task::JoinHandle<()> {
+    tokio::task::spawn(async move {
+        loop {
+            let state_guard = state.lock().await;
+            let sweep_addresses = state_guard.sweep_addresses.clone();
+            drop(state_guard);
+            for (id, sweep_address_task) in sweep_addresses.iter() {
+                if let SweepAddressAddendum::Monero(addendum) = sweep_address_task.addendum.clone()
+                {
+                    let mut sweep_address_txs = None;
+                    match sweep_address(
+                        monero::Address::from_str(&addendum.address).unwrap(),
+                        addendum.view_key,
+                        addendum.spend_key,
+                        &network,
+                        Arc::clone(&wallet),
+                    )
+                    .await
+                    {
+                        Ok(val) => {
+                            sweep_address_txs = val;
+                        }
+                        Err(err) => {
+                            trace!("error polling sweep address {:?}, retrying", err);
+                        }
+                    }
+                    if let Some(txids) = sweep_address_txs {
+                        let mut state_guard = state.lock().await;
+                        state_guard.success_sweep(id, txids).await;
+                        drop(state_guard);
+                    }
+                }
+            }
+            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+        }
+    })
+}
+
 fn unseen_transaction_polling(
     state: Arc<Mutex<SyncerState>>,
     syncer_servers: SyncerServers,
@@ -473,6 +606,18 @@ impl Synclet for MoneroSyncer {
         if !polling {
             error!("monero syncer only supports polling for now - switching to polling=true");
         }
+        let network = match chain.clone() {
+            Chain::Mainnet | Chain::Regtest(_) => monero::Network::Mainnet,
+            Chain::Testnet3 => monero::Network::Stagenet,
+            Chain::Signet => monero::Network::Testnet,
+            _ => {
+                error!(
+                    "invalid chain type for monero: {}- switching to mainnet",
+                    chain
+                );
+                monero::Network::Mainnet
+            }
+        };
         let _handle = std::thread::spawn(move || {
             use tokio::runtime::Builder;
             let rt = Builder::new_multi_thread()
@@ -481,6 +626,9 @@ impl Synclet for MoneroSyncer {
                 .build()
                 .unwrap();
             rt.block_on(async {
+                let wallet_mutex = Arc::new(Mutex::new(
+                    monero_rpc::RpcClient::new(syncer_servers.monero_rpc_wallet.clone()).wallet(),
+                ));
                 let (event_tx, event_rx): (
                     TokioSender<SyncerdBridgeEvent>,
                     TokioReceiver<SyncerdBridgeEvent>,
@@ -495,21 +643,12 @@ impl Synclet for MoneroSyncer {
                 .await;
                 run_syncerd_bridge_event_sender(tx, event_rx, syncer_address).await;
 
-                let network = match chain.clone() {
-                    Chain::Mainnet | Chain::Regtest(_) => monero::Network::Mainnet,
-                    Chain::Testnet3 => monero::Network::Stagenet,
-                    Chain::Signet => monero::Network::Testnet,
-                    _ => {
-                        error!(
-                            "invalid chain type for monero: {}- switching to mainnet",
-                            chain
-                        );
-                        monero::Network::Mainnet
-                    }
-                };
-
-                let address_handle =
-                    address_polling(Arc::clone(&state), syncer_servers.clone(), network);
+                let address_handle = address_polling(
+                    Arc::clone(&state),
+                    syncer_servers.clone(),
+                    network.clone(),
+                    Arc::clone(&wallet_mutex),
+                );
 
                 // transaction polling is done in the same loop
                 let height_handle = height_polling(Arc::clone(&state), syncer_servers.clone());
@@ -517,8 +656,15 @@ impl Synclet for MoneroSyncer {
                 let unseen_transaction_handle =
                     unseen_transaction_polling(Arc::clone(&state), syncer_servers.clone());
 
-                let res =
-                    tokio::try_join!(address_handle, height_handle, unseen_transaction_handle);
+                let sweep_handle =
+                    sweep_polling(Arc::clone(&state), Arc::clone(&wallet_mutex), network);
+
+                let res = tokio::try_join!(
+                    address_handle,
+                    height_handle,
+                    unseen_transaction_handle,
+                    sweep_handle
+                );
                 debug!("exiting monero synclet run routine with: {:?}", res);
             });
         });
diff --git a/src/syncerd/opts.rs b/src/syncerd/opts.rs
index 6cf57413..3b8162b8 100644
--- a/src/syncerd/opts.rs
+++ b/src/syncerd/opts.rs
@@ -13,6 +13,7 @@
 // If not, see <https://opensource.org/licenses/MIT>.
 
 use clap::{AppSettings, Clap};
+use farcaster_core::blockchain;
 use std::str::FromStr;
 use strict_encoding::{StrictDecode, StrictEncode};
 
@@ -82,3 +83,11 @@ impl Opts {
         self.shared.process();
     }
 }
+
+// fn into(network: &farcaster_core::blockchain::Network) -> monero::Network {
+//     match network {
+//         blockchain::Network::Mainnet => monero::Network::Mainnet,
+//         blockchain::Network::Testnet => monero::Network::Stagenet,
+//         blockchain::Network::Local => monero::Network::Mainnet,
+//     }
+// }
diff --git a/src/syncerd/syncer_state.rs b/src/syncerd/syncer_state.rs
index 887494c7..79a82ca9 100644
--- a/src/syncerd/syncer_state.rs
+++ b/src/syncerd/syncer_state.rs
@@ -1,4 +1,4 @@
-use crate::farcaster_core::consensus::Decodable;
+use crate::farcaster_core::{blockchain::Network, consensus::Decodable};
 use crate::rpc::request::SyncerdBridgeEvent;
 use crate::syncerd::opts::Coin;
 use crate::syncerd::runtime::SyncerdTask;
@@ -25,6 +25,7 @@ pub struct SyncerState {
     pub addresses: HashMap<u32, AddressTransactions>,
     pub transactions: HashMap<u32, WatchedTransaction>,
     pub unseen_transactions: HashSet<u32>,
+    pub sweep_addresses: HashMap<u32, SweepAddress>,
     tx_event: TokioSender<SyncerdBridgeEvent>,
     task_count: u32,
 }
@@ -63,6 +64,7 @@ impl SyncerState {
             addresses: HashMap::new(),
             transactions: HashMap::new(),
             unseen_transactions: HashSet::new(),
+            sweep_addresses: HashMap::new(),
             tx_event,
             task_count: 0,
         }
@@ -75,9 +77,9 @@ impl SyncerState {
             .iter()
             .filter_map(|(id, address_transaction)| {
                 if address_transaction.task.id == task_id {
-                    return Some(id.clone());
+                    Some(id.clone())
                 } else {
-                    return None;
+                    None
                 }
             })
             .collect();
@@ -107,9 +109,9 @@ impl SyncerState {
             .iter()
             .filter_map(|(id, watched_transaction)| {
                 if watched_transaction.task.id == task_id {
-                    return Some(id.clone());
+                    Some(id.clone())
                 } else {
-                    return None;
+                    None
                 }
             })
             .collect();
@@ -139,9 +141,9 @@ impl SyncerState {
             .iter()
             .filter_map(|(id, watch_height)| {
                 if watch_height.id == task_id {
-                    return Some(id.clone());
+                    Some(id.clone())
                 } else {
-                    return None;
+                    None
                 }
             })
             .collect();
@@ -165,6 +167,38 @@ impl SyncerState {
             }
         }
 
+        // check sweep address tasks
+        let ids: Vec<u32> = self
+            .sweep_addresses
+            .iter()
+            .filter_map(|(id, sweep_address)| {
+                if sweep_address.id == task_id {
+                    Some(id.clone())
+                } else {
+                    None
+                }
+            })
+            .collect();
+        for id in ids.iter() {
+            if let Some(source_id) = self.tasks_sources.get(&id) {
+                if *source_id == source {
+                    self.remove_sweep_address(id);
+                    send_event(
+                        &self.tx_event,
+                        &mut vec![(
+                            Event::TaskAborted(TaskAborted {
+                                id: task_id.clone(),
+                                error: None,
+                            }),
+                            source.clone(),
+                        )],
+                    )
+                    .await;
+                    return;
+                }
+            }
+        }
+
         send_event(
             &self.tx_event,
             &mut vec![(
@@ -245,6 +279,18 @@ impl SyncerState {
         self.unseen_transactions.insert(self.task_count);
     }
 
+    pub fn sweep_address(&mut self, task: SweepAddress, source: ServiceId) {
+        self.task_count += 1;
+        // This is technically valid behavior; immediately prune the task for being past
+        // its lifetime by never inserting it
+        if let Err(e) = self.add_lifetime(task.lifetime, self.task_count) {
+            error!("{}", e);
+            return;
+        }
+        self.sweep_addresses.insert(self.task_count, task.clone());
+        self.tasks_sources.insert(self.task_count, source.clone());
+    }
+
     pub async fn change_height(&mut self, height: u64, block: Vec<u8>) {
         if self.block_height != height || self.block_hash != block {
             self.block_height = height;
@@ -394,9 +440,7 @@ impl SyncerState {
                     if confirmations.is_some() {
                         unseen_transactions.remove(&id);
                     } else {
-                        if !unseen_transactions.contains(&id) {
-                            unseen_transactions.insert(id.clone());
-                        }
+                        unseen_transactions.insert(id.clone());
                     }
                     let transaction_confirmations = if confirmations
                         != watched_tx.transaction_confirmations.confirmations
@@ -434,6 +478,33 @@ impl SyncerState {
         send_event(&self.tx_event, &mut events).await;
     }
 
+    pub async fn success_sweep(&mut self, id: &u32, txids: Vec<Vec<u8>>) {
+        if let Some(sweep_address) = self.sweep_addresses.get(id) {
+            send_event(
+                &self.tx_event,
+                &mut vec![(
+                    Event::SweepSuccess(SweepSuccess {
+                        id: sweep_address.id,
+                        txids,
+                    }),
+                    self.tasks_sources
+                        .get(&id)
+                        .cloned()
+                        .expect("task source missing"),
+                )],
+            )
+            .await;
+            if let Some(ids) = self.lifetimes.get_mut(&sweep_address.lifetime) {
+                ids.remove(id);
+                if ids.is_empty() {
+                    self.lifetimes.remove(&sweep_address.lifetime);
+                }
+            }
+            self.sweep_addresses.remove(id);
+            self.tasks_sources.remove(id);
+        }
+    }
+
     fn drop_lifetimes(&mut self) {
         let lifetimes: Vec<u64> = Iterator::collect(self.lifetimes.keys().map(|&x| x.to_owned()));
         for lifetime in lifetimes {
@@ -466,6 +537,7 @@ impl SyncerState {
                 self.transactions.remove(task);
                 self.unseen_transactions.remove(task);
                 self.watch_height.remove(task);
+                self.sweep_addresses.remove(task);
                 self.tasks_sources.remove(task);
             }
         } else {
@@ -512,6 +584,19 @@ impl SyncerState {
         self.watch_height.remove(id);
         self.tasks_sources.remove(id);
     }
+
+    fn remove_sweep_address(&mut self, id: &u32) {
+        if let Some(sweep_address) = self.sweep_addresses.get(id) {
+            if let Some(ids) = self.lifetimes.get_mut(&sweep_address.lifetime) {
+                ids.remove(id);
+                if ids.is_empty() {
+                    self.lifetimes.remove(&sweep_address.lifetime);
+                }
+            }
+        }
+        self.sweep_addresses.remove(id);
+        self.tasks_sources.remove(id);
+    }
 }
 
 async fn send_event(
@@ -548,25 +633,15 @@ async fn syncer_state_transaction() {
         confirmation_bound: 4,
     };
     let height_task = WatchHeight { id: 0, lifetime: 4 };
+    let source1 = ServiceId::Syncer(Coin::Bitcoin, Network::Mainnet);
 
-    state.watch_transaction(
-        transaction_task_one.clone(),
-        ServiceId::Syncer(Coin::Bitcoin),
-    );
-    state.abort(0, ServiceId::Syncer(Coin::Bitcoin)).await;
+    state.watch_transaction(transaction_task_one.clone(), source1.clone());
+    state.abort(0, source1.clone()).await;
     assert!(event_rx.try_recv().is_ok());
 
-    state.watch_transaction(
-        transaction_task_one.clone(),
-        ServiceId::Syncer(Coin::Bitcoin),
-    );
-    state.watch_transaction(
-        transaction_task_two.clone(),
-        ServiceId::Syncer(Coin::Bitcoin),
-    );
-    state
-        .watch_height(height_task, ServiceId::Syncer(Coin::Bitcoin))
-        .await;
+    state.watch_transaction(transaction_task_one.clone(), source1.clone());
+    state.watch_transaction(transaction_task_two.clone(), source1.clone());
+    state.watch_height(height_task, source1.clone()).await;
     assert_eq!(state.lifetimes.len(), 3);
     assert_eq!(state.transactions.len(), 2);
     assert_eq!(state.tasks_sources.len(), 3);
@@ -619,11 +694,9 @@ async fn syncer_state_transaction() {
     assert_eq!(state.unseen_transactions.len(), 2);
     assert!(event_rx.try_recv().is_ok());
 
-    state.watch_transaction(
-        transaction_task_two.clone(),
-        ServiceId::Syncer(Coin::Monero),
-    );
-    state.abort(0, ServiceId::Syncer(Coin::Monero)).await;
+    let source2 = ServiceId::Syncer(Coin::Monero, Network::Mainnet);
+    state.watch_transaction(transaction_task_two.clone(), source2.clone());
+    state.abort(0, source2.clone()).await;
     assert_eq!(state.lifetimes.len(), 3);
     assert_eq!(state.transactions.len(), 2);
     assert_eq!(state.tasks_sources.len(), 3);
@@ -665,19 +738,18 @@ async fn syncer_state_addresses() {
         addendum: addendum.clone(),
         include_tx: Boolean::False,
     };
+    let source1 = ServiceId::Syncer(Coin::Bitcoin, Network::Mainnet);
 
     state
-        .watch_address(address_task_two.clone(), ServiceId::Syncer(Coin::Bitcoin))
+        .watch_address(address_task_two.clone(), source1.clone())
         .unwrap();
-    state.abort(0, ServiceId::Syncer(Coin::Bitcoin)).await;
+    state.abort(0, source1.clone()).await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.addresses.len(), 0);
     assert!(event_rx.try_recv().is_ok());
 
-    state
-        .watch_address(address_task, ServiceId::Syncer(Coin::Bitcoin))
-        .unwrap();
+    state.watch_address(address_task, source1.clone()).unwrap();
     assert_eq!(state.lifetimes.len(), 1);
     assert_eq!(state.tasks_sources.len(), 1);
     assert_eq!(state.addresses.len(), 1);
@@ -763,10 +835,11 @@ async fn syncer_state_addresses() {
     assert!(event_rx.try_recv().is_ok());
     assert!(event_rx.try_recv().is_ok());
 
+    let source2 = ServiceId::Syncer(Coin::Monero, Network::Testnet);
     state
-        .watch_address(address_task_two.clone(), ServiceId::Syncer(Coin::Monero))
+        .watch_address(address_task_two.clone(), source2.clone())
         .unwrap();
-    state.abort(0, ServiceId::Syncer(Coin::Monero)).await;
+    state.abort(0, source2.clone()).await;
     assert_eq!(state.lifetimes.len(), 1);
     assert_eq!(state.tasks_sources.len(), 1);
     assert_eq!(state.addresses.len(), 1);
@@ -774,9 +847,7 @@ async fn syncer_state_addresses() {
 
     state.change_height(1, vec![0]).await;
     let height_task = WatchHeight { id: 0, lifetime: 3 };
-    state
-        .watch_height(height_task, ServiceId::Syncer(Coin::Bitcoin))
-        .await;
+    state.watch_height(height_task, source1.clone()).await;
     assert_eq!(state.lifetimes.len(), 2);
     assert_eq!(state.tasks_sources.len(), 2);
     assert!(event_rx.try_recv().is_ok());
@@ -794,6 +865,53 @@ async fn syncer_state_addresses() {
     assert!(event_rx.try_recv().is_ok());
 }
 
+#[tokio::test]
+async fn syncer_state_sweep_addresses() {
+    use std::str::FromStr;
+    use tokio::sync::mpsc::Receiver as TokioReceiver;
+    let (event_tx, mut event_rx): (
+        TokioSender<SyncerdBridgeEvent>,
+        TokioReceiver<SyncerdBridgeEvent>,
+    ) = tokio::sync::mpsc::channel(120);
+    let mut state = SyncerState::new(event_tx.clone());
+    let sweep_task = SweepAddress {
+        id: 0,
+        lifetime: 11,
+        addendum: SweepAddressAddendum::Monero(SweepXmrAddress {
+            view_key: monero::PrivateKey::from_str(
+                "77916d0cd56ed1920aef6ca56d8a41bac915b68e4c46a589e0956e27a7b77404",
+            )
+            .unwrap(),
+            spend_key: monero::PrivateKey::from_str(
+                "77916d0cd56ed1920aef6ca56d8a41bac915b68e4c46a589e0956e27a7b77404",
+            )
+            .unwrap(),
+            address: "address".to_string(),
+        }),
+    };
+    let source1 = ServiceId::Syncer(Coin::Monero, Network::Mainnet);
+
+    state.sweep_address(sweep_task.clone(), source1.clone());
+    assert_eq!(state.lifetimes.len(), 1);
+    assert_eq!(state.tasks_sources.len(), 1);
+    assert_eq!(state.sweep_addresses.len(), 1);
+    state.abort(0, source1.clone()).await;
+    assert_eq!(state.lifetimes.len(), 0);
+    assert_eq!(state.tasks_sources.len(), 0);
+    assert_eq!(state.sweep_addresses.len(), 0);
+    assert!(event_rx.try_recv().is_ok());
+
+    state.sweep_address(sweep_task, source1.clone());
+    assert_eq!(state.lifetimes.len(), 1);
+    assert_eq!(state.tasks_sources.len(), 1);
+    assert_eq!(state.sweep_addresses.len(), 1);
+    state.success_sweep(&2, vec![vec![0]]).await;
+    assert_eq!(state.lifetimes.len(), 0);
+    assert_eq!(state.tasks_sources.len(), 0);
+    assert_eq!(state.sweep_addresses.len(), 0);
+    assert!(event_rx.try_recv().is_ok());
+}
+
 #[tokio::test]
 async fn syncer_state_height() {
     use tokio::sync::mpsc::Receiver as TokioReceiver;
@@ -804,24 +922,22 @@ async fn syncer_state_height() {
     let mut state = SyncerState::new(event_tx.clone());
     let height_task = WatchHeight { id: 0, lifetime: 0 };
     let another_height_task = WatchHeight { id: 0, lifetime: 3 };
+    let source1 = ServiceId::Syncer(Coin::Bitcoin, Network::Mainnet);
 
     state
-        .watch_height(height_task.clone(), ServiceId::Syncer(Coin::Bitcoin))
+        .watch_height(height_task.clone(), source1.clone())
         .await;
-    state.abort(0, ServiceId::Syncer(Coin::Bitcoin)).await;
+    state.abort(0, source1.clone()).await;
     assert_eq!(state.lifetimes.len(), 0);
     assert_eq!(state.tasks_sources.len(), 0);
     assert_eq!(state.watch_height.len(), 0);
     assert!(event_rx.try_recv().is_ok());
 
     state
-        .watch_height(height_task.clone(), ServiceId::Syncer(Coin::Bitcoin))
+        .watch_height(height_task.clone(), source1.clone())
         .await;
     state
-        .watch_height(
-            another_height_task.clone(),
-            ServiceId::Syncer(Coin::Bitcoin),
-        )
+        .watch_height(another_height_task.clone(), source1.clone())
         .await;
     assert_eq!(state.lifetimes.len(), 2);
     assert_eq!(state.tasks_sources.len(), 2);
@@ -846,10 +962,11 @@ async fn syncer_state_height() {
     assert_eq!(state.watch_height.len(), 1);
     assert!(event_rx.try_recv().is_err());
 
+    let source2 = ServiceId::Syncer(Coin::Monero, Network::Mainnet);
     state
-        .watch_height(another_height_task.clone(), ServiceId::Syncer(Coin::Monero))
+        .watch_height(another_height_task.clone(), source2.clone())
         .await;
-    state.abort(0, ServiceId::Syncer(Coin::Monero)).await;
+    state.abort(0, source2.clone()).await;
     assert_eq!(state.lifetimes.len(), 1);
     assert_eq!(state.tasks_sources.len(), 1);
     assert_eq!(state.watch_height.len(), 1);
diff --git a/src/syncerd/types.rs b/src/syncerd/types.rs
index 78c12460..20c24076 100644
--- a/src/syncerd/types.rs
+++ b/src/syncerd/types.rs
@@ -26,6 +26,36 @@ pub struct XmrAddressAddendum {
     pub from_height: u64,
 }
 
+#[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
+#[display(Debug)]
+pub struct SweepAddress {
+    pub id: u32,
+    pub lifetime: u64,
+    pub addendum: SweepAddressAddendum,
+}
+
+#[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
+#[display(Debug)]
+pub enum SweepAddressAddendum {
+    Monero(SweepXmrAddress),
+    Bitcoin(SweepBitcoinAddress),
+}
+
+#[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
+#[display(Debug)]
+pub struct SweepXmrAddress {
+    pub spend_key: monero::PrivateKey,
+    pub view_key: monero::PrivateKey,
+    pub address: String,
+}
+
+#[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
+#[display(Debug)]
+pub struct SweepBitcoinAddress {
+    pub private_key: [u8; 32],
+    pub address: bitcoin::Address,
+}
+
 #[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
 #[display(Debug)]
 pub struct Abort {
@@ -90,6 +120,7 @@ pub enum Task {
     WatchAddress(WatchAddress),
     WatchTransaction(WatchTransaction),
     BroadcastTransaction(BroadcastTransaction),
+    SweepAddress(SweepAddress),
 }
 
 #[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
@@ -133,6 +164,13 @@ pub struct TransactionBroadcasted {
     pub error: Option<String>,
 }
 
+#[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
+#[display(Debug)]
+pub struct SweepSuccess {
+    pub id: u32,
+    pub txids: Vec<Vec<u8>>,
+}
+
 /// Events returned by syncers to the daemon to update the blockchain states.
 /// Events are identified with a unique 32-bits integer that match the [`Task`]
 /// id.
@@ -144,6 +182,7 @@ pub enum Event {
     AddressTransaction(AddressTransaction),
     TransactionConfirmations(TransactionConfirmations),
     TransactionBroadcasted(TransactionBroadcasted),
+    SweepSuccess(SweepSuccess),
     /// Notify the daemon the task has been aborted with success or failure.
     /// Carries the status for the task abortion.
     TaskAborted(TaskAborted),
diff --git a/src/walletd/runtime.rs b/src/walletd/runtime.rs
index 0feae3d9..95a02e4d 100644
--- a/src/walletd/runtime.rs
+++ b/src/walletd/runtime.rs
@@ -7,14 +7,17 @@ use std::{
     str::FromStr,
 };
 
-use crate::rpc::{
-    request::{self, BitcoinAddress, Commit, Keys, Msg, Params, Reveal, Token, Tx},
-    Request, ServiceBus,
-};
 use crate::swapd::get_swap_id;
 use crate::walletd::NodeSecrets;
 use crate::LogStyle;
 use crate::Senders;
+use crate::{
+    rpc::{
+        request::{self, BitcoinAddress, Commit, Keys, Msg, Params, Reveal, Token, Tx},
+        Request, ServiceBus,
+    },
+    syncerd::SweepXmrAddress,
+};
 use crate::{Config, CtlServer, Error, Service, ServiceId};
 use bitcoin::{
     hashes::hex::FromHex,
@@ -1098,6 +1101,7 @@ impl Runtime {
                     remote_params: Some(alice_params),
                     core_arb_setup: Some(_),
                     adaptor_buy: Some(adaptor_buy),
+                    pub_offer,
                     ..
                 })) = self.wallets.get_mut(&get_swap_id(&source)?)
                 {
@@ -1112,9 +1116,9 @@ impl Runtime {
                     let sk_a = monero::PrivateKey::from_slice(sk_a_btc_buf.as_ref())
                         .expect("Valid Monero Private Key");
                     info!("Extracted alice's monero key from Buy tx: {}", sk_a);
-
                     let sk_b = key_manager.get_or_derive_monero_spend_key()?;
-                    info!("Full secret monero spending key: {}", sk_a + sk_b);
+                    let spend_key = sk_a + sk_b;
+                    info!("Full secret monero spending key: {}", spend_key);
 
                     let view_key = *alice_params
                         .accordant_shared_keys
@@ -1123,7 +1127,20 @@ impl Runtime {
                         .find(|vk| vk.tag() == &SharedKeyId::new(SHARED_VIEW_KEY_ID))
                         .unwrap()
                         .elem();
-                    info!("Full secret monero view key: {}", view_key);
+                    info!("Full secret monero viewkey: {}", view_key);
+
+                    let address = s!("76KwxdgtWyJQK5a27PpqT973R6s7bZ4cKWTtUG6JEyrbLQy5F4ZwdUqgKpssGQzRxnd99LKWqFjYo92b8ngS7GqD1VZBEEL");
+                    let sweep_keys = SweepXmrAddress {
+                        view_key,
+                        spend_key,
+                        address,
+                    };
+                    senders.send_to(
+                        ServiceBus::Ctl,
+                        self.identity(),
+                        source,
+                        Request::SweepXmrAddress(sweep_keys),
+                    )?;
                 }
             }
             Request::Tx(Tx::Refund(refund_tx)) => {
@@ -1149,7 +1166,8 @@ impl Runtime {
                     info!("Extracted alice's monero key from Buy tx: {}", sk_b);
 
                     let sk_c = key_manager.get_or_derive_monero_spend_key()?;
-                    info!("Full secret monero spending key: {}", sk_b + sk_c);
+                    let spend_key = sk_b + sk_c;
+                    info!("Full secret monero spending key: {}", spend_key);
 
                     let view_key = *bob_params
                         .accordant_shared_keys
@@ -1159,6 +1177,18 @@ impl Runtime {
                         .unwrap()
                         .elem();
                     info!("Full secret monero view key: {}", view_key);
+                    let address = s!("73oh6eWaf9MStZuxXqeE4nVx9jP5Y8XBZeLQ9AbAqFMWfGLSxjPdkbigWsb2PYvzyTWN3p2wa1jJn28V3vb4zqA4UoCXJp9");
+                    let sweep_keys = SweepXmrAddress {
+                        view_key,
+                        spend_key,
+                        address,
+                    };
+                    senders.send_to(
+                        ServiceBus::Ctl,
+                        self.identity(),
+                        source,
+                        Request::SweepXmrAddress(sweep_keys),
+                    )?;
                 }
             }
             Request::GetKeys(request::GetKeys(wallet_token, request_id)) => {


Test patch:
diff --git a/tests/functional.rs b/tests/functional.rs
index 66418515..bd4dcc64 100644
--- a/tests/functional.rs
+++ b/tests/functional.rs
@@ -6,6 +6,9 @@ use farcaster_node::syncerd::bitcoin_syncer::Synclet;
 use farcaster_node::syncerd::monero_syncer::MoneroSyncer;
 use farcaster_node::syncerd::opts::Coin;
 use farcaster_node::syncerd::runtime::SyncerdTask;
+use farcaster_node::syncerd::SweepAddress;
+use farcaster_node::syncerd::SweepAddressAddendum;
+use farcaster_node::syncerd::SweepXmrAddress;
 use farcaster_node::syncerd::SyncerServers;
 use farcaster_node::syncerd::XmrAddressAddendum;
 use farcaster_node::ServiceId;
@@ -25,11 +28,15 @@ use bitcoin::hashes::Hash;
 use internet2::{CreateUnmarshaller, Unmarshall};
 use std::str::FromStr;
 
+use farcaster_core::blockchain::Network;
 use farcaster_node::syncerd::types::{
     Abort, AddressAddendum, Boolean, BroadcastTransaction, BtcAddressAddendum, Event, Task,
     WatchAddress, WatchHeight, WatchTransaction,
 };
 
+const SOURCE1: ServiceId = ServiceId::Syncer(Coin::Bitcoin, Network::Local);
+const SOURCE2: ServiceId = ServiceId::Syncer(Coin::Monero, Network::Local);
+
 /*
 These tests need to run serialy, otherwise we cannot verify events based on the
 state of electrum and bitcoin
@@ -52,6 +59,7 @@ fn bitcoin_syncer_test() {
 #[tokio::test]
 #[ignore] // it's too expensive
 async fn monero_syncer_test() {
+    monero_syncer_sweep_test().await;
     monero_syncer_block_height_test().await;
     monero_syncer_address_test().await;
     monero_syncer_transaction_test().await;
@@ -85,7 +93,7 @@ fn bitcoin_syncer_block_height_test(polling: bool) {
             id: 0,
             lifetime: blocks + 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
 
@@ -111,7 +119,7 @@ fn bitcoin_syncer_block_height_test(polling: bool) {
             id: 1,
             lifetime: blocks + 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
     let message = rx_event.recv_multipart(0).unwrap();
@@ -192,7 +200,7 @@ fn bitcoin_syncer_address_test(polling: bool) {
             addendum: addendum_1,
             include_tx: Boolean::True,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(watch_address_task_1).unwrap();
     let watch_address_task_2 = SyncerdTask {
@@ -202,7 +210,7 @@ fn bitcoin_syncer_address_test(polling: bool) {
             addendum: addendum_2.clone(),
             include_tx: Boolean::True,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(watch_address_task_2).unwrap();
 
@@ -266,7 +274,7 @@ fn bitcoin_syncer_address_test(polling: bool) {
             addendum: addendum_2,
             include_tx: Boolean::True,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(watch_address_task_3).unwrap();
     println!("waiting for watch transaction message");
@@ -302,7 +310,7 @@ fn bitcoin_syncer_address_test(polling: bool) {
                 addendum: addendum_4.clone(),
                 include_tx: Boolean::True,
             }),
-            source: ServiceId::Syncer(Coin::Bitcoin),
+            source: SOURCE1.clone(),
         })
         .unwrap();
     }
@@ -359,7 +367,7 @@ fn bitcoin_syncer_transaction_test(polling: bool) {
             hash: txid_1.to_vec(),
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     })
     .unwrap();
 
@@ -367,21 +375,21 @@ fn bitcoin_syncer_transaction_test(polling: bool) {
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 0, vec![0]);
+    assert_transaction_confirmations(request, Some(0), vec![0]);
 
     let block_hash = bitcoin_rpc.generate_to_address(1, &address).unwrap();
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 1, block_hash[0].to_vec());
+    assert_transaction_confirmations(request, Some(1), block_hash[0].to_vec());
 
     bitcoin_rpc.generate_to_address(1, &address).unwrap();
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 2, block_hash[0].to_vec());
+    assert_transaction_confirmations(request, Some(2), block_hash[0].to_vec());
 
     let block_hash = bitcoin_rpc.generate_to_address(1, &address).unwrap();
     let block = bitcoin_rpc.get_block(&block_hash[0]).unwrap();
@@ -393,21 +401,21 @@ fn bitcoin_syncer_transaction_test(polling: bool) {
             hash: address_txid.to_vec(),
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     })
     .unwrap();
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 1, block_hash[0].to_vec());
+    assert_transaction_confirmations(request, Some(1), block_hash[0].to_vec());
 
     bitcoin_rpc.generate_to_address(1, &address).unwrap();
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 2, block_hash[0].to_vec());
+    assert_transaction_confirmations(request, Some(2), block_hash[0].to_vec());
 
     let txid_2 = bitcoin_rpc
         .send_to_address(&address, amount, None, None, None, None, None, None)
@@ -423,7 +431,7 @@ fn bitcoin_syncer_transaction_test(polling: bool) {
             hash: txid_2.to_vec(),
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     })
     .unwrap();
     tx.send(SyncerdTask {
@@ -433,7 +441,7 @@ fn bitcoin_syncer_transaction_test(polling: bool) {
             hash: txid_3.to_vec(),
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     })
     .unwrap();
 
@@ -441,12 +449,12 @@ fn bitcoin_syncer_transaction_test(polling: bool) {
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 0, vec![0]);
+    assert_transaction_confirmations(request, Some(0), vec![0]);
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 0, vec![0]);
+    assert_transaction_confirmations(request, Some(0), vec![0]);
 }
 
 fn bitcoin_syncer_abort_test() {
@@ -461,13 +469,13 @@ fn bitcoin_syncer_abort_test() {
             hash: vec![0; 32],
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
 
     let task = SyncerdTask {
         task: Task::Abort(Abort { id: 0 }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
     println!("waiting for task aborted message");
@@ -478,7 +486,7 @@ fn bitcoin_syncer_abort_test() {
 
     let task = SyncerdTask {
         task: Task::Abort(Abort { id: 0 }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
     println!("waiting for task aborted message");
@@ -526,7 +534,7 @@ fn bitcoin_syncer_broadcast_tx_test(polling: bool) {
             id: 0,
             tx: transaction.hex,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
 
@@ -571,7 +579,7 @@ fn bitcoin_syncer_broadcast_tx_test(polling: bool) {
             id: 0,
             tx: signed_tx.hex,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
 
@@ -598,7 +606,7 @@ fn create_bitcoin_syncer(polling: bool) -> (std::sync::mpsc::Sender<SyncerdTask>
     syncer.run(
         rx,
         tx_event,
-        ServiceId::Syncer(Coin::Bitcoin).into(),
+        SOURCE1.clone().into(),
         syncer_servers,
         Chain::Regtest(dummy),
         polling,
@@ -671,7 +679,7 @@ async fn monero_syncer_block_height_test() {
             id: 0,
             lifetime: blocks + 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
 
@@ -696,7 +704,7 @@ async fn monero_syncer_block_height_test() {
             id: 1,
             lifetime: u64::from(blocks) + 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(task).unwrap();
     println!("waiting for height changed");
@@ -719,6 +727,55 @@ async fn monero_syncer_block_height_test() {
     assert_received_height_changed(request, blocks);
 }
 
+async fn monero_syncer_sweep_test() {
+    let (regtest, wallet) = setup_monero().await;
+    let address = wallet.get_address(0, None).await.unwrap();
+    let blocks = regtest.generate_blocks(200, address.address).await.unwrap();
+
+    let duration = std::time::Duration::from_secs(20);
+    std::thread::sleep(duration);
+
+    let (tx, rx_event) = create_monero_syncer();
+
+    let spend_key = monero::PrivateKey::from_str(
+        "77916d0cd56ed1920aef6ca56d8a41bac915b68e4c46a589e0956e27a7b77404",
+    )
+    .unwrap();
+    let view_key = monero::PrivateKey::from_str(
+        "8163466f1883598e6dd14027b8da727057165da91485834314f5500a65846f09",
+    )
+    .unwrap();
+    let keypair = monero::KeyPair {
+        view: view_key,
+        spend: spend_key,
+    };
+    let to_be_sweeped_address = monero::Address::from_keypair(monero::Network::Mainnet, &keypair);
+    let dest_address = "43qHP7gSJJf8HZw1G3ZmpWVyYnbxkKdfta34Qj2nuRENjAsXBtj9JcMWcYMeT3n4NyTZqxhUkKgsTS6P2TNgM6ksM32czSp";
+    send_monero(&wallet, to_be_sweeped_address, 1000000000000).await;
+
+    let task = SyncerdTask {
+        task: Task::SweepAddress(SweepAddress {
+            id: 0,
+            lifetime: blocks + 40,
+            addendum: SweepAddressAddendum::Monero(SweepXmrAddress {
+                spend_key,
+                view_key,
+                address: dest_address.to_string(),
+            }),
+        }),
+        source: SOURCE2.clone(),
+    };
+    tx.send(task).unwrap();
+
+    regtest.generate_blocks(20, address.address).await.unwrap();
+
+    println!("waiting for sweep address message");
+    let message = rx_event.recv_multipart(0).unwrap();
+    println!("received sweep success message");
+    let request = get_request_from_message(message);
+    println!("request: {:?}", request);
+}
+
 /*
 We test for the following scenarios in the address transaction tests:
 
@@ -763,7 +820,7 @@ async fn monero_syncer_address_test() {
             addendum: addendum_1,
             include_tx: Boolean::True,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(watch_address_task_1).unwrap();
 
@@ -790,7 +847,7 @@ async fn monero_syncer_address_test() {
             addendum: addendum_2,
             include_tx: Boolean::True,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(watch_address_task_2).unwrap();
 
@@ -818,7 +875,7 @@ async fn monero_syncer_address_test() {
             addendum: addendum_3,
             include_tx: Boolean::True,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     };
     tx.send(watch_address_task_3).unwrap();
     println!("waiting for address transaction message");
@@ -849,7 +906,7 @@ async fn monero_syncer_address_test() {
                 addendum: addendum_4.clone(),
                 include_tx: Boolean::True,
             }),
-            source: ServiceId::Syncer(Coin::Monero),
+            source: SOURCE2.clone(),
         })
         .unwrap();
     }
@@ -894,7 +951,7 @@ async fn monero_syncer_transaction_test() {
             hash: txid_1.to_vec(),
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     })
     .unwrap();
 
@@ -902,7 +959,7 @@ async fn monero_syncer_transaction_test() {
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 0, vec![0]);
+    assert_transaction_confirmations(request, Some(0), vec![0]);
 
     let block_height = regtest.generate_blocks(1, address).await.unwrap();
     let block_hash = get_block_hash_from_height(&regtest, block_height).await;
@@ -910,14 +967,14 @@ async fn monero_syncer_transaction_test() {
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 1, block_hash.clone());
+    assert_transaction_confirmations(request, Some(1), block_hash.clone());
 
     regtest.generate_blocks(1, address).await.unwrap();
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 2, block_hash);
+    assert_transaction_confirmations(request, Some(2), block_hash);
 
     let tx_id2 = send_monero(&wallet, address, 1).await;
     let block_height = regtest.generate_blocks(1, address).await.unwrap();
@@ -929,21 +986,21 @@ async fn monero_syncer_transaction_test() {
             hash: tx_id2,
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     })
     .unwrap();
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 1, block_hash.clone());
+    assert_transaction_confirmations(request, Some(1), block_hash.clone());
 
     regtest.generate_blocks(1, address).await.unwrap();
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 2, block_hash);
+    assert_transaction_confirmations(request, Some(2), block_hash);
 
     let txid_2 = send_monero(&wallet, address, 1).await;
     let txid_3 = send_monero(&wallet, address, 1).await;
@@ -955,7 +1012,7 @@ async fn monero_syncer_transaction_test() {
             hash: txid_2.to_vec(),
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     })
     .unwrap();
     tx.send(SyncerdTask {
@@ -965,7 +1022,7 @@ async fn monero_syncer_transaction_test() {
             hash: txid_3.to_vec(),
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Bitcoin),
+        source: SOURCE1.clone(),
     })
     .unwrap();
 
@@ -973,12 +1030,12 @@ async fn monero_syncer_transaction_test() {
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 0, vec![0]);
+    assert_transaction_confirmations(request, Some(0), vec![0]);
     println!("awaiting confirmations");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("received confirmation");
     let request = get_request_from_message(message);
-    assert_transaction_confirmations(request, 0, vec![0]);
+    assert_transaction_confirmations(request, Some(0), vec![0]);
 }
 
 async fn monero_syncer_abort_test() {
@@ -994,24 +1051,28 @@ async fn monero_syncer_abort_test() {
             hash: vec![0; 32],
             confirmation_bound: 2,
         }),
-        source: ServiceId::Syncer(Coin::Monero),
+        source: SOURCE2.clone(),
     };
     tx.send(task).unwrap();
+    let message = rx_event.recv_multipart(0).unwrap();
+    let request = get_request_from_message(message);
+    assert_transaction_confirmations(request, None, vec![0]);
 
     let task = SyncerdTask {
         task: Task::Abort(Abort { id: 0 }),
-        source: ServiceId::Syncer(Coin::Monero),
+        source: SOURCE2.clone(),
     };
     tx.send(task).unwrap();
     println!("waiting for task aborted message");
     let message = rx_event.recv_multipart(0).unwrap();
     println!("task aborted");
     let request = get_request_from_message(message);
+    println!("message: {:?}", request);
     assert_task_aborted(request, None, 0);
 
     let task = SyncerdTask {
         task: Task::Abort(Abort { id: 0 }),
-        source: ServiceId::Syncer(Coin::Monero),
+        source: SOURCE2.clone(),
     };
     tx.send(task).unwrap();
     println!("waiting for task aborted message");
@@ -1037,7 +1098,7 @@ async fn monero_syncer_broadcast_tx_test() {
 
     let task = SyncerdTask {
         task: Task::BroadcastTransaction(BroadcastTransaction { id: 0, tx: vec![0] }),
-        source: ServiceId::Syncer(Coin::Monero),
+        source: SOURCE2.clone(),
     };
     tx.send(task).unwrap();
 
@@ -1085,7 +1146,7 @@ fn create_monero_syncer() -> (std::sync::mpsc::Sender<SyncerdTask>, zmq::Socket)
     syncer.run(
         rx,
         tx_event,
-        ServiceId::Syncer(Coin::Monero).into(),
+        SOURCE2.clone().into(),
         syncer_servers,
         Chain::Regtest(dummy),
         true,
@@ -1195,7 +1256,7 @@ fn assert_received_height_changed(request: Request, expected_height: u64) {
 
 fn assert_transaction_confirmations(
     request: Request,
-    expected_confirmations: u32,
+    expected_confirmations: Option<u32>,
     expected_block_hash: Vec<u8>,
 ) {
     match request {
@@ -1203,7 +1264,7 @@ fn assert_transaction_confirmations(
             Event::TransactionConfirmations(transaction_confirmations) => {
                 assert_eq!(
                     transaction_confirmations.confirmations,
-                    Some(expected_confirmations)
+                    expected_confirmations
                 );
                 assert_eq!(transaction_confirmations.block, expected_block_hash);
             }