Repository: farcaster-project/farcaster-node
Pull request #555 (2022-07-11 22:46:55)

farcasterd: launch syncers before launching swapd, not after


Patch:
diff --git a/src/farcasterd/runtime.rs b/src/farcasterd/runtime.rs
index af63af37..9782d506 100644
--- a/src/farcasterd/runtime.rs
+++ b/src/farcasterd/runtime.rs
@@ -608,10 +608,10 @@ impl Runtime {
                             &source,
                             self.syncer_services.len().bright_blue_bold()
                         );
-                        for (source, request) in self.pending_sweep_requests.drain() {
+                        for (_, request) in self.pending_sweep_requests.drain() {
                             endpoints.send_to(
                                 ServiceBus::Ctl,
-                                source,
+                                ServiceId::Farcasterd,
                                 ServiceId::Syncer(*coin, *network),
                                 request,
                             )?;
@@ -629,7 +629,7 @@ impl Runtime {
                     }
                 };
 
-                if let Some((swap_params, network)) = self.making_swaps.get(&source) {
+                if let Some((swap_params, _network)) = self.making_swaps.get(&source) {
                     // Tell swapd swap options and link it with the
                     // connection daemon
                     debug!(
@@ -644,29 +644,6 @@ impl Runtime {
                             source
                         ))),
                     ));
-                    let swapid = get_swap_id(&source)?;
-                    // when online, Syncers say Hello, then they get registered to self.syncers
-                    syncers_up(
-                        ServiceId::Farcasterd,
-                        &mut self.spawning_services,
-                        &self.syncer_services,
-                        &mut self.syncer_clients,
-                        Coin::Bitcoin,
-                        *network,
-                        swapid,
-                        &self.config,
-                    )?;
-                    syncers_up(
-                        ServiceId::Farcasterd,
-                        &mut self.spawning_services,
-                        &self.syncer_services,
-                        &mut self.syncer_clients,
-                        Coin::Monero,
-                        *network,
-                        swapid,
-                        &self.config,
-                    )?;
-                    // FIXME msgs should go to walletd?
                     endpoints.send_to(
                         ServiceBus::Ctl,
                         self.identity(),
@@ -675,7 +652,7 @@ impl Runtime {
                     )?;
                     self.running_swaps.insert(swap_params.swap_id);
                     self.making_swaps.remove(&source);
-                } else if let Some((swap_params, network)) = self.taking_swaps.get(&source) {
+                } else if let Some((swap_params, _network)) = self.taking_swaps.get(&source) {
                     // Tell swapd swap options and link it with the
                     // connection daemon
                     debug!(
@@ -690,33 +667,6 @@ impl Runtime {
                             source
                         ))),
                     ));
-                    match swap_params.local_params {
-                        Params::Alice(_) => {}
-                        Params::Bob(_) => {}
-                    }
-
-                    let swapid = get_swap_id(&source)?;
-                    syncers_up(
-                        ServiceId::Farcasterd,
-                        &mut self.spawning_services,
-                        &self.syncer_services,
-                        &mut self.syncer_clients,
-                        Coin::Bitcoin,
-                        *network,
-                        swapid,
-                        &self.config,
-                    )?;
-                    syncers_up(
-                        ServiceId::Farcasterd,
-                        &mut self.spawning_services,
-                        &self.syncer_services,
-                        &mut self.syncer_clients,
-                        Coin::Monero,
-                        *network,
-                        swapid,
-                        &self.config,
-                    )?;
-                    // FIXME msgs should go to walletd?
                     endpoints.send_to(
                         ServiceBus::Ctl,
                         self.identity(),
@@ -790,6 +740,7 @@ impl Runtime {
                 funding_address,
             }) => {
                 let offerid = public_offer.offer.id();
+                let network = public_offer.offer.network.clone();
                 let peerd_id = self.peerd_ids.get(&offerid); // Some for Maker after TakerCommit, None for Taker
                 let peer: ServiceId = match local_trade_role {
                     TradeRole::Maker if peerd_id.is_some() => peerd_id.unwrap().clone(),
@@ -815,7 +766,31 @@ impl Runtime {
                             status: OfferStatus::InProgress,
                         }),
                     )?;
+                    syncers_up(
+                        ServiceId::Farcasterd,
+                        &mut self.spawning_services,
+                        &self.syncer_services,
+                        &mut self.syncer_clients,
+                        Coin::Bitcoin,
+                        network,
+                        swap_id,
+                        &self.config,
+                    )?;
+                    syncers_up(
+                        ServiceId::Farcasterd,
+                        &mut self.spawning_services,
+                        &self.syncer_services,
+                        &mut self.syncer_clients,
+                        Coin::Monero,
+                        network,
+                        swap_id,
+                        &self.config,
+                    )?;
 
+                    if !self.syncer_services.contains_key(&(Coin::Bitcoin, network)) {
+                        trace!("give time for syncer to come online");
+                        std::thread::sleep(Duration::from_secs_f32(10.0));
+                    }
                     trace!(
                         "launching swapd with swap_id: {}",
                         swap_id.bright_yellow_bold()
@@ -1683,7 +1658,6 @@ impl Runtime {
                 let id = TaskId(self.syncer_task_counter);
                 let request = Request::SyncerTask(Task::SweepAddress(SweepAddress {
                     id,
-                    retry: false,
                     lifetime: u64::MAX,
                     addendum: SweepAddressAddendum::Bitcoin(sweep_bitcoin_address),
                     from_height: None,
@@ -1715,7 +1689,7 @@ impl Runtime {
                 } else {
                     endpoints.send_to(
                         ServiceBus::Ctl,
-                        source,
+                        ServiceId::Farcasterd,
                         ServiceId::Syncer(coin, network),
                         request,
                     )?;
diff --git a/src/peerd/runtime.rs b/src/peerd/runtime.rs
index 52bff0b4..9d249e7e 100644
--- a/src/peerd/runtime.rs
+++ b/src/peerd/runtime.rs
@@ -27,8 +27,7 @@ use bitcoin::secp256k1::rand::{self, Rng, RngCore};
 use bitcoin::secp256k1::PublicKey;
 use internet2::{addr::InetSocketAddr, CreateUnmarshaller, Unmarshall, Unmarshaller};
 use internet2::{presentation, transport, zmqsocket, NodeAddr, TypedEnum, ZmqType, ZMQ_CONTEXT};
-
-use crate::rpc::messages::Ping;
+use lnp::p2p::legacy::{Messages, Ping};
 use microservices::esb::{self, Handler};
 use microservices::node::TryService;
 use microservices::peer::{self, PeerConnection, PeerSender, SendMessage};
diff --git a/src/rpc/request.rs b/src/rpc/request.rs
index 2bf0e73b..6cf7a660 100644
--- a/src/rpc/request.rs
+++ b/src/rpc/request.rs
@@ -15,7 +15,6 @@
 #![allow(clippy::clone_on_copy)]
 
 use crate::cli::OfferSelector;
-use crate::rpc::messages::Ping;
 use crate::swapd::CheckpointSwapd;
 use crate::syncerd::SweepBitcoinAddress;
 use crate::walletd::{
@@ -79,6 +78,7 @@ use farcaster_core::{
 };
 use internet2::Api;
 use internet2::{NodeAddr, RemoteSocketAddr};
+use lnp::p2p::legacy::{Messages, Ping, TempChannelId as TempSwapId};
 use microservices::rpc::Failure;
 use microservices::rpc_connection;
 use strict_encoding::{
@@ -379,6 +379,10 @@ pub enum Request {
     #[display("hello()")]
     Hello,
 
+    #[api(type = 2)]
+    #[display("send_message({0})")]
+    PeerMessage(Messages),
+
     #[api(type = 3)]
     #[display("terminate")]
     Terminate,
diff --git a/src/swapd/runtime.rs b/src/swapd/runtime.rs
index d2ec1937..7aa6dd9d 100644
--- a/src/swapd/runtime.rs
+++ b/src/swapd/runtime.rs
@@ -491,7 +491,8 @@ impl Runtime {
                 "Incorrect peer connection", self.peer_service, source
             )));
         }
-        let msg = match &request {
+        let msg_bus = ServiceBus::Msg;
+        match &request {
             Request::Protocol(msg) => {
                 if msg.swap_id() != self.swap_id() {
                     return Err(Error::Farcaster(format!(
@@ -500,345 +501,347 @@ impl Runtime {
                         self.swap_id(),
                         msg.swap_id(),
                     )));
-                } else {
-                    msg
                 }
-            }
-            _ => {
-                error!("MSG RPC can be only used for forwarding farcaster protocol messages");
-                return Err(Error::NotSupported(ServiceBus::Msg, request.get_type()));
-            }
-        };
-        let msg_bus = ServiceBus::Msg;
-        match &msg {
-            // we are taker and the maker committed, now we reveal after checking
-            // whether we're Bob or Alice and that we're on a compatible state
-            Msg::MakerCommit(remote_commit)
-                if self.state.commit()
-                    && self.state.trade_role() == Some(TradeRole::Taker)
-                    && self.state.remote_commit().is_none() =>
-            {
-                trace!("received remote commitment");
-                self.state.t_sup_remote_commit(remote_commit.clone());
+                match &msg {
+                    // we are taker and the maker committed, now we reveal after checking
+                    // whether we're Bob or Alice and that we're on a compatible state
+                    Msg::MakerCommit(remote_commit)
+                        if self.state.commit()
+                            && self.state.trade_role() == Some(TradeRole::Taker)
+                            && self.state.remote_commit().is_none() =>
+                    {
+                        trace!("received remote commitment");
+                        self.state.t_sup_remote_commit(remote_commit.clone());
 
-                let watch_height_btc_task = self.syncer_state.watch_height(Coin::Bitcoin);
-                endpoints.send_to(
-                    ServiceBus::Ctl,
-                    self.identity(),
-                    self.syncer_state.bitcoin_syncer(),
-                    Request::SyncerTask(watch_height_btc_task),
-                )?;
+                        let watch_height_btc_task = self.syncer_state.watch_height(Coin::Bitcoin);
+                        endpoints.send_to(
+                            ServiceBus::Ctl,
+                            self.identity(),
+                            self.syncer_state.bitcoin_syncer(),
+                            Request::SyncerTask(watch_height_btc_task),
+                        )?;
 
-                let watch_height_xmr_task = self.syncer_state.watch_height(Coin::Monero);
-                endpoints.send_to(
-                    ServiceBus::Ctl,
-                    self.identity(),
-                    self.syncer_state.monero_syncer(),
-                    Request::SyncerTask(watch_height_xmr_task),
-                )?;
+                        let watch_height_xmr_task = self.syncer_state.watch_height(Coin::Monero);
 
-                if self.state.swap_role() == SwapRole::Bob {
-                    let addr = self
-                        .state
-                        .b_address()
-                        .cloned()
-                        .expect("address available at CommitB");
-                    let txlabel = TxLabel::Funding;
-                    if !self.syncer_state.is_watched_addr(&txlabel) {
-                        let task = self
-                            .syncer_state
-                            .watch_addr_btc(addr.script_pubkey(), txlabel);
-                        self.send_ctl(
-                            endpoints,
-                            self.syncer_state.bitcoin_syncer(),
-                            Request::SyncerTask(task),
+                        endpoints.send_to(
+                            ServiceBus::Ctl,
+                            self.identity(),
+                            self.syncer_state.monero_syncer(),
+                            Request::SyncerTask(watch_height_xmr_task),
                         )?;
-                    }
-                    let btc_fee_task = self.syncer_state.estimate_fee_btc();
-                    endpoints.send_to(
-                        ServiceBus::Ctl,
-                        self.identity(),
-                        self.syncer_state.bitcoin_syncer(),
-                        Request::SyncerTask(btc_fee_task),
-                    )?;
-                    std::thread::sleep(Duration::from_secs_f32(2.0));
-                }
 
-                self.send_wallet(msg_bus, endpoints, request)?;
-            }
-            Msg::TakerCommit(_) => {
-                unreachable!(
-                    "msg handled by farcasterd/walletd, and indirectly here by \
-                             Ctl Request::MakeSwap"
-                )
-            }
-            Msg::Reveal(Reveal::Proof(_)) => {
-                // These messages are saved as pending if Bob and then forwarded once the
-                // parameter reveal forward is triggered. If Alice, send immediately.
-                match self.state.swap_role() {
-                    SwapRole::Bob => {
-                        let pending_request = PendingRequest {
-                            request,
-                            dest: ServiceId::Wallet,
-                            bus_id: ServiceBus::Msg,
-                        };
-                        trace!("Added pending request to be forwarded later",);
-                        if self
-                            .pending_requests
-                            .insert(ServiceId::Wallet, vec![pending_request])
-                            .is_some()
-                        {
-                            error!("Pending requests already existed prior to Reveal::Proof!")
+                        if self.state.swap_role() == SwapRole::Bob {
+                            let addr = self
+                                .state
+                                .b_address()
+                                .cloned()
+                                .expect("address available at CommitB");
+                            let txlabel = TxLabel::Funding;
+                            if !self.syncer_state.is_watched_addr(&txlabel) {
+                                let task = self
+                                    .syncer_state
+                                    .watch_addr_btc(addr.script_pubkey(), txlabel);
+                                self.send_ctl(
+                                    endpoints,
+                                    self.syncer_state.bitcoin_syncer(),
+                                    Request::SyncerTask(task),
+                                )?;
+                            }
+                            let btc_fee_task = self.syncer_state.estimate_fee_btc();
+                            endpoints.send_to(
+                                ServiceBus::Ctl,
+                                self.identity(),
+                                self.syncer_state.bitcoin_syncer(),
+                                Request::SyncerTask(btc_fee_task),
+                            )?;
+                            std::thread::sleep(Duration::from_secs_f32(2.0));
                         }
+
+                        self.send_wallet(msg_bus, endpoints, request)?;
                     }
-                    SwapRole::Alice => {
-                        debug!("Alice: forwarding reveal");
-                        trace!(
-                            "sending request {} to {} on bus {}",
-                            &request,
-                            &ServiceId::Wallet,
-                            &ServiceBus::Msg
-                        );
-                        self.send_wallet(msg_bus, endpoints, request)?
+                    Msg::TakerCommit(_) => {
+                        unreachable!(
+                            "msg handled by farcasterd/walletd, and indirectly here by \
+                             Ctl Request::MakeSwap"
+                        )
                     }
-                }
-            }
-            // bob and alice
-            // store parameters from counterparty if we have not received them yet.
-            // if we're maker, also reveal to taker if their commitment is valid.
-            Msg::Reveal(reveal)
-                if self.state.remote_commit().is_some()
-                    && (self.state.commit() || self.state.reveal()) =>
-            {
-                // TODO: since we're not actually revealing, find other name for
-                // intermediary state
+                    Msg::Reveal(Reveal::Proof(_)) => {
+                        // These messages are saved as pending if Bob and then forwarded once the
+                        // parameter reveal forward is triggered. If Alice, send immediately.
+                        match self.state.swap_role() {
+                            SwapRole::Bob => {
+                                let pending_request = PendingRequest {
+                                    request,
+                                    dest: ServiceId::Wallet,
+                                    bus_id: ServiceBus::Msg,
+                                };
+                                trace!("Added pending request to be forwarded later",);
+                                if self
+                                    .pending_requests
+                                    .insert(ServiceId::Wallet, vec![pending_request])
+                                    .is_some()
+                                {
+                                    error!(
+                                        "Pending requests already existed prior to Reveal::Proof!"
+                                    )
+                                }
+                            }
+                            SwapRole::Alice => {
+                                debug!("Alice: forwarding reveal");
+                                trace!(
+                                    "sending request {} to {} on bus {}",
+                                    &request,
+                                    &ServiceId::Wallet,
+                                    &ServiceBus::Msg
+                                );
+                                self.send_wallet(msg_bus, endpoints, request)?
+                            }
+                        }
+                    }
+                    // bob and alice
+                    // store parameters from counterparty if we have not received them yet.
+                    // if we're maker, also reveal to taker if their commitment is valid.
+                    Msg::Reveal(reveal)
+                        if self.state.remote_commit().is_some()
+                            && (self.state.commit() || self.state.reveal()) =>
+                    {
+                        // TODO: since we're not actually revealing, find other name for
+                        // intermediary state
 
-                let remote_commit = self.state.remote_commit().cloned().unwrap();
+                        let remote_commit = self.state.remote_commit().cloned().unwrap();
 
-                if let Ok(remote_params_candidate) = remote_params_candidate(reveal, remote_commit)
-                {
-                    debug!("{:?} sets remote_params", self.state.swap_role());
-                    self.state.sup_remote_params(remote_params_candidate);
-                } else {
-                    error!("Revealed remote params not preimage of commitment");
-                }
+                        if let Ok(remote_params_candidate) =
+                            remote_params_candidate(reveal, remote_commit)
+                        {
+                            debug!("{:?} sets remote_params", self.state.swap_role());
+                            self.state.sup_remote_params(remote_params_candidate);
+                        } else {
+                            error!("Revealed remote params not preimage of commitment");
+                        }
 
-                // Specific to swap roles
-                // pass request on to wallet daemon so that it can set remote params
-                match self.state.swap_role() {
-                    // validated state above, no need to check again
-                    SwapRole::Alice => {
-                        // Alice already sends RevealProof immediately, so only have to
-                        // forward Reveal now
-                        trace!(
-                            "sending request {} to {} on bus {}",
-                            &request,
-                            &ServiceId::Wallet,
-                            &ServiceBus::Msg
-                        );
-                        self.send_wallet(msg_bus, endpoints, request)?
-                    }
-                    SwapRole::Bob => {
-                        // sending this request will initialize the
-                        // arbitrating setup, that can be only performed
-                        // after the funding tx was seen
-                        let pending_request = PendingRequest {
-                            request,
-                            dest: ServiceId::Wallet,
-                            bus_id: ServiceBus::Msg,
-                        };
+                        // Specific to swap roles
+                        // pass request on to wallet daemon so that it can set remote params
+                        match self.state.swap_role() {
+                            // validated state above, no need to check again
+                            SwapRole::Alice => {
+                                // Alice already sends RevealProof immediately, so only have to
+                                // forward Reveal now
+                                trace!(
+                                    "sending request {} to {} on bus {}",
+                                    &request,
+                                    &ServiceId::Wallet,
+                                    &ServiceBus::Msg
+                                );
+                                self.send_wallet(msg_bus, endpoints, request)?
+                            }
+                            SwapRole::Bob => {
+                                // sending this request will initialize the
+                                // arbitrating setup, that can be only performed
+                                // after the funding tx was seen
+                                let pending_request = PendingRequest {
+                                    request,
+                                    dest: ServiceId::Wallet,
+                                    bus_id: ServiceBus::Msg,
+                                };
+
+                                trace!(
+                                    "This pending request will be called later: {:?}",
+                                    &pending_request
+                                );
+                                let pending_requests = self.pending_requests.get_mut(&ServiceId::Wallet)
+                                    .expect("should already have received Reveal::Proof, so this key should exist.");
+                                if pending_requests.len() != 1 {
+                                    error!("should have a single pending Reveal::Proof only FIXME")
+                                }
+                                pending_requests.push(pending_request);
 
-                        trace!(
-                            "This pending request will be called later: {:?}",
-                            &pending_request
-                        );
-                        let pending_requests = self
-                            .pending_requests
-                            .get_mut(&ServiceId::Wallet)
-                            .expect(
-                            "should already have received Reveal::Proof, so this key should exist.",
-                        );
-                        if pending_requests.len() != 1 {
-                            error!("should have a single pending Reveal::Proof only FIXME")
-                        }
-                        pending_requests.push(pending_request);
-
-                        if let (Some(address), Some(sat_per_kvb)) = (
-                            self.state.b_address().cloned(),
-                            self.syncer_state.btc_fee_estimate_sat_per_kvb,
-                        ) {
-                            let swap_id = self.swap_id();
-                            let vsize = 94;
-                            let nr_inputs = 1;
-                            let total_fees = bitcoin::Amount::from_sat(p2wpkh_signed_tx_fee(
-                                sat_per_kvb,
-                                vsize,
-                                nr_inputs,
-                            ));
-                            let amount = self.syncer_state.bitcoin_amount + total_fees;
-                            info!(
+                                if let (Some(address), Some(sat_per_kvb)) = (
+                                    self.state.b_address().cloned(),
+                                    self.syncer_state.btc_fee_estimate_sat_per_kvb,
+                                ) {
+                                    let swap_id = self.swap_id();
+                                    let vsize = 94;
+                                    let nr_inputs = 1;
+                                    let total_fees = bitcoin::Amount::from_sat(
+                                        p2wpkh_signed_tx_fee(sat_per_kvb, vsize, nr_inputs),
+                                    );
+                                    let amount = self.syncer_state.bitcoin_amount + total_fees;
+                                    info!(
                                         "{} | Send {} to {}, this includes {} for the Lock transaction network fees",
                                         swap_id.bright_blue_italic(),
                                         amount.bright_green_bold(),
                                         address.addr(),
                                         total_fees,
                                     );
-                            self.state.b_sup_required_funding_amount(amount);
-                            let req =
-                                Request::FundingInfo(FundingInfo::Bitcoin(BitcoinFundingInfo {
-                                    swap_id,
-                                    address,
-                                    amount,
-                                }));
-                            self.syncer_state.awaiting_funding = true;
-
-                            if let Some(enquirer) = self.enquirer.clone() {
+                                    self.state.b_sup_required_funding_amount(amount);
+                                    let req = Request::FundingInfo(FundingInfo::Bitcoin(
+                                        BitcoinFundingInfo {
+                                            swap_id,
+                                            address,
+                                            amount,
+                                        },
+                                    ));
+                                    self.syncer_state.awaiting_funding = true;
+
+                                    if let Some(enquirer) = self.enquirer.clone() {
+                                        endpoints.send_to(
+                                            ServiceBus::Ctl,
+                                            self.identity(),
+                                            enquirer,
+                                            req,
+                                        )?
+                                    }
+                                } else {
+                                    error!(
+                                        "swap_role: {}, trade_role: {:?}",
+                                        self.state.swap_role(),
+                                        self.state.trade_role()
+                                    );
+                                    error!("Not Some(address) or not Some(sat_per_kvb)");
+                                }
+                            }
+                        }
+
+                        // up to here for both maker and taker, following only Maker
+
+                        // if did not yet reveal, maker only. on the msg flow as
+                        // of 2021-07-13 taker reveals first
+                        if self.state.commit() && self.state.trade_role() == Some(TradeRole::Maker)
+                        {
+                            if let Some(addr) = self.state.b_address().cloned() {
+                                let txlabel = TxLabel::Funding;
+                                if !self.syncer_state.is_watched_addr(&txlabel) {
+                                    let watch_addr_task = self
+                                        .syncer_state
+                                        .watch_addr_btc(addr.script_pubkey(), txlabel);
+                                    self.send_ctl(
+                                        endpoints,
+                                        self.syncer_state.bitcoin_syncer(),
+                                        Request::SyncerTask(watch_addr_task),
+                                    )?;
+                                }
+                            }
+                            let watch_height_btc = self.syncer_state.watch_height(Coin::Bitcoin);
+                            endpoints.send_to(
+                                ServiceBus::Ctl,
+                                self.identity(),
+                                self.syncer_state.bitcoin_syncer(),
+                                Request::SyncerTask(watch_height_btc),
+                            )?;
+
+                            let watch_height_xmr = self.syncer_state.watch_height(Coin::Monero);
+                            endpoints.send_to(
+                                ServiceBus::Ctl,
+                                self.identity(),
+                                self.syncer_state.monero_syncer(),
+                                Request::SyncerTask(watch_height_xmr),
+                            )?;
+                        }
+                    }
+                    // alice receives, bob sends
+                    Msg::CoreArbitratingSetup(CoreArbitratingSetup {
+                        swap_id,
+                        lock,
+                        cancel,
+                        refund,
+                        ..
+                    }) if self.state.swap_role() == SwapRole::Alice
+                        && self.state.reveal()
+                        && swap_id == &self.swap_id() =>
+                    {
+                        for (&tx, tx_label) in [lock, cancel, refund].iter().zip([
+                            TxLabel::Lock,
+                            TxLabel::Cancel,
+                            TxLabel::Refund,
+                        ]) {
+                            let tx = tx.clone().extract_tx();
+                            let txid = tx.txid();
+                            debug!(
+                                "tx_label: {}, vsize: {}, outs: {}",
+                                tx_label,
+                                tx.vsize(),
+                                tx.output.len()
+                            );
+                            if !self.syncer_state.is_watched_tx(&tx_label) {
+                                let task = self.syncer_state.watch_tx_btc(txid, tx_label);
                                 endpoints.send_to(
                                     ServiceBus::Ctl,
                                     self.identity(),
-                                    enquirer,
-                                    req,
-                                )?
+                                    self.syncer_state.bitcoin_syncer(),
+                                    Request::SyncerTask(task),
+                                )?;
+                            }
+                            if tx_label == TxLabel::Refund {
+                                self.syncer_state.tasks.txids.insert(TxLabel::Refund, txid);
                             }
-                        } else {
-                            error!(
-                                "swap_role: {}, trade_role: {:?}",
-                                self.state.swap_role(),
-                                self.state.trade_role()
-                            );
-                            error!("Not Some(address) or not Some(sat_per_kvb)");
                         }
+                        self.send_wallet(msg_bus, endpoints, request)?;
                     }
-                }
+                    // bob receives, alice sends
+                    Msg::RefundProcedureSignatures(_) if self.state.b_core_arb() => {
+                        self.state.sup_received_refund_procedure_signatures();
+                        self.send_wallet(msg_bus, endpoints, request)?;
+                    }
+                    // alice receives, bob sends
+                    Msg::BuyProcedureSignature(buy_proc_sig)
+                        if self.state.a_refundsig() && !self.state.a_overfunded() =>
+                    {
+                        // Alice verifies that she has sent refund procedure signatures before
+                        // processing the buy signatures from Bob
+                        let tx_label = TxLabel::Buy;
+                        if !self.syncer_state.is_watched_tx(&tx_label) {
+                            let txid = buy_proc_sig.buy.clone().extract_tx().txid();
+                            let task = self.syncer_state.watch_tx_btc(txid, tx_label);
+                            endpoints.send_to(
+                                ServiceBus::Ctl,
+                                self.identity(),
+                                self.syncer_state.bitcoin_syncer(),
+                                Request::SyncerTask(task),
+                            )?;
+                        }
 
-                // up to here for both maker and taker, following only Maker
-
-                // if did not yet reveal, maker only. on the msg flow as
-                // of 2021-07-13 taker reveals first
-                if self.state.commit() && self.state.trade_role() == Some(TradeRole::Maker) {
-                    if let Some(addr) = self.state.b_address().cloned() {
-                        let txlabel = TxLabel::Funding;
-                        if !self.syncer_state.is_watched_addr(&txlabel) {
-                            let watch_addr_task = self
-                                .syncer_state
-                                .watch_addr_btc(addr.script_pubkey(), txlabel);
-                            self.send_ctl(
+                        // checkpoint swap alice pre buy
+                        debug!(
+                            "{} | checkpointing alice pre buy swapd state",
+                            self.swap_id.bright_blue_italic()
+                        );
+                        if self.state.a_sup_checkpoint_pre_buy() {
+                            checkpoint_send(
                                 endpoints,
-                                self.syncer_state.bitcoin_syncer(),
-                                Request::SyncerTask(watch_addr_task),
+                                self.swap_id,
+                                self.identity(),
+                                ServiceId::Database,
+                                request::CheckpointState::CheckpointSwapd(CheckpointSwapd {
+                                    state: self.state.clone(),
+                                    last_msg: Msg::BuyProcedureSignature(buy_proc_sig.clone()),
+                                    enquirer: self.enquirer.clone(),
+                                    temporal_safety: self.temporal_safety.clone(),
+                                    txs: self.txs.clone(),
+                                    txids: self.syncer_state.tasks.txids.clone(),
+                                    pending_requests: self.pending_requests.clone(),
+                                    pending_broadcasts: self.syncer_state.pending_broadcast_txs(),
+                                    xmr_addr_addendum: self.syncer_state.xmr_addr_addendum.clone(),
+                                }),
                             )?;
                         }
+
+                        self.send_wallet(msg_bus, endpoints, request)?
                     }
-                    let watch_height_btc = self.syncer_state.watch_height(Coin::Bitcoin);
-                    endpoints.send_to(
-                        ServiceBus::Ctl,
-                        self.identity(),
-                        self.syncer_state.bitcoin_syncer(),
-                        Request::SyncerTask(watch_height_btc),
-                    )?;
 
-                    let watch_height_xmr = self.syncer_state.watch_height(Coin::Monero);
-                    endpoints.send_to(
-                        ServiceBus::Ctl,
-                        self.identity(),
-                        self.syncer_state.monero_syncer(),
-                        Request::SyncerTask(watch_height_xmr),
-                    )?;
-                }
-            }
-            // alice receives, bob sends
-            Msg::CoreArbitratingSetup(CoreArbitratingSetup {
-                swap_id,
-                lock,
-                cancel,
-                refund,
-                ..
-            }) if self.state.swap_role() == SwapRole::Alice && self.state.reveal() => {
-                for (&tx, tx_label) in [lock, cancel, refund].iter().zip([
-                    TxLabel::Lock,
-                    TxLabel::Cancel,
-                    TxLabel::Refund,
-                ]) {
-                    let tx = tx.clone().extract_tx();
-                    let txid = tx.txid();
-                    debug!(
-                        "tx_label: {}, vsize: {}, outs: {}",
-                        tx_label,
-                        tx.vsize(),
-                        tx.output.len()
-                    );
-                    if !self.syncer_state.is_watched_tx(&tx_label) {
-                        let task = self.syncer_state.watch_tx_btc(txid, tx_label);
-                        endpoints.send_to(
-                            ServiceBus::Ctl,
-                            self.identity(),
-                            self.syncer_state.bitcoin_syncer(),
-                            Request::SyncerTask(task),
-                        )?;
+                    // bob and alice
+                    Msg::Abort(_) => {
+                        return Err(Error::Farcaster("Abort not yet supported".to_string()))
                     }
-                    if tx_label == TxLabel::Refund {
-                        self.syncer_state.tasks.txids.insert(TxLabel::Refund, txid);
+                    Msg::Ping(_) | Msg::Pong(_) | Msg::PingPeer => {
+                        unreachable!("ping/pong must remain in peerd, and unreachable in swapd")
                     }
+                    request => error!("request not supported {}", request),
                 }
-                self.send_wallet(msg_bus, endpoints, request)?;
             }
-            // bob receives, alice sends
-            Msg::RefundProcedureSignatures(_) if self.state.b_core_arb() => {
-                self.state.sup_received_refund_procedure_signatures();
-                self.send_wallet(msg_bus, endpoints, request)?;
-            }
-            // alice receives, bob sends
-            Msg::BuyProcedureSignature(buy_proc_sig)
-                if self.state.a_refundsig() && !self.state.a_overfunded() =>
-            {
-                // Alice verifies that she has sent refund procedure signatures before
-                // processing the buy signatures from Bob
-                let tx_label = TxLabel::Buy;
-                if !self.syncer_state.is_watched_tx(&tx_label) {
-                    let txid = buy_proc_sig.buy.clone().extract_tx().txid();
-                    let task = self.syncer_state.watch_tx_btc(txid, tx_label);
-                    endpoints.send_to(
-                        ServiceBus::Ctl,
-                        self.identity(),
-                        self.syncer_state.bitcoin_syncer(),
-                        Request::SyncerTask(task),
-                    )?;
-                }
-
-                // checkpoint swap alice pre buy
-                debug!(
-                    "{} | checkpointing alice pre buy swapd state",
-                    self.swap_id.bright_blue_italic()
-                );
-                if self.state.a_sup_checkpoint_pre_buy() {
-                    checkpoint_send(
-                        endpoints,
-                        self.swap_id,
-                        self.identity(),
-                        ServiceId::Database,
-                        request::CheckpointState::CheckpointSwapd(CheckpointSwapd {
-                            state: self.state.clone(),
-                            last_msg: Msg::BuyProcedureSignature(buy_proc_sig.clone()),
-                            enquirer: self.enquirer.clone(),
-                            temporal_safety: self.temporal_safety.clone(),
-                            txs: self.txs.clone(),
-                            txids: self.syncer_state.tasks.txids.clone(),
-                            pending_requests: self.pending_requests.clone(),
-                            pending_broadcasts: self.syncer_state.pending_broadcast_txs(),
-                            xmr_addr_addendum: self.syncer_state.xmr_addr_addendum.clone(),
-                        }),
-                    )?;
-                }
-
-                self.send_wallet(msg_bus, endpoints, request)?
-            }
-
-            // bob and alice
-            Msg::Abort(_) => return Err(Error::Farcaster("Abort not yet supported".to_string())),
-            Msg::Ping(_) | Msg::Pong(_) | Msg::PingPeer => {
-                unreachable!("ping/pong must remain in peerd, and unreachable in swapd")
+            _ => {
+                error!("MSG RPC can be only used for forwarding farcaster protocol messages");
+                return Err(Error::NotSupported(ServiceBus::Msg, request.get_type()));
             }
-            request => error!("request not supported {}", request),
         }
         Ok(())
     }
@@ -898,7 +901,6 @@ impl Runtime {
                     dest_address,
                     from_height,
                     minimum_balance,
-                    true,
                 );
                 let acc_confs_needs =
                     self.temporal_safety.sweep_monero_thr - self.temporal_safety.xmr_finality_thr;
@@ -1034,7 +1036,6 @@ impl Runtime {
                     Some(remote_commit),
                 );
 
-                std::thread::sleep(Duration::from_secs_f32(5.0));
                 let btc_fee_task = self.syncer_state.estimate_fee_btc();
                 endpoints.send_to(
                     ServiceBus::Ctl,
@@ -2156,7 +2157,7 @@ impl Runtime {
                     sweep_bitcoin_address.destination_address
                 );
 
-                let task = self.syncer_state.sweep_btc(sweep_bitcoin_address, false);
+                let task = self.syncer_state.sweep_btc(sweep_bitcoin_address);
                 endpoints.send_to(
                     ServiceBus::Ctl,
                     self.identity(),
diff --git a/src/swapd/syncer_client.rs b/src/swapd/syncer_client.rs
index 773f28f1..e779ac7a 100644
--- a/src/swapd/syncer_client.rs
+++ b/src/swapd/syncer_client.rs
@@ -254,7 +254,6 @@ impl SyncerState {
         dest_address: monero::Address,
         from_height: Option<u64>,
         minimum_balance: monero::Amount,
-        retry: bool,
     ) -> Task {
         let id = self.tasks.new_taskid();
         self.tasks.sweeping_addr = Some(id);
@@ -269,7 +268,6 @@ impl SyncerState {
             id,
             lifetime,
             addendum,
-            retry,
             from_height,
         };
         let task = Task::SweepAddress(sweep_task);
@@ -287,7 +285,7 @@ impl SyncerState {
         self.tasks.tasks.insert(id, task.clone());
         task
     }
-    pub fn sweep_btc(&mut self, sweep_bitcoin_address: SweepBitcoinAddress, retry: bool) -> Task {
+    pub fn sweep_btc(&mut self, sweep_bitcoin_address: SweepBitcoinAddress) -> Task {
         let id = self.tasks.new_taskid();
         self.tasks.sweeping_addr = Some(id);
         let lifetime = self.task_lifetime(Coin::Bitcoin);
@@ -296,7 +294,6 @@ impl SyncerState {
             id,
             lifetime,
             addendum,
-            retry,
             from_height: None,
         };
         let task = Task::SweepAddress(sweep_task);
diff --git a/src/syncerd/bitcoin_syncer.rs b/src/syncerd/bitcoin_syncer.rs
index e42b6b82..9c5f0825 100644
--- a/src/syncerd/bitcoin_syncer.rs
+++ b/src/syncerd/bitcoin_syncer.rs
@@ -878,11 +878,12 @@ fn estimate_fee_polling(
                 loop {
                     match client
                         .estimate_fee(high_priority_confs) // docs say sat/kB, but its BTC/kvB
-                        .and_then(|high_priority| {
-                            client
-                                .estimate_fee(low_priority_confs)
-                                .map(|low_priority| (high_priority, low_priority))
-                        }) {
+                        .and_then(
+                            |high_priority| match client.estimate_fee(low_priority_confs) {
+                                Ok(low_priority) => Ok((high_priority, low_priority)),
+                                Err(err) => Err(err),
+                            },
+                        ) {
                         Ok((high_priority, low_priority)) => {
                             let mut state_guard = state.lock().await;
                             state_guard
@@ -944,11 +945,7 @@ fn sweep_polling(
                                 });
                                 debug!("sweep address transaction: {:?}", sweep_address_txs);
                                 let mut state_guard = state.lock().await;
-                                if !sweep_address_txs.is_empty() {
-                                    state_guard.success_sweep(id, sweep_address_txs).await;
-                                } else if !sweep_address_task.retry {
-                                    state_guard.fail_sweep(id).await;
-                                }
+                                state_guard.success_sweep(id, sweep_address_txs).await;
                                 drop(state_guard);
                             } else {
                                 error!("Not sweeping address - is not using a bitcoin sweep address addendum");
diff --git a/src/syncerd/monero_syncer.rs b/src/syncerd/monero_syncer.rs
index e792b500..e0f5c4df 100644
--- a/src/syncerd/monero_syncer.rs
+++ b/src/syncerd/monero_syncer.rs
@@ -678,13 +678,11 @@ fn sweep_polling(
                         warn!("error polling sweep address {:?}, retrying", err);
                         vec![]
                     });
-                    let mut state_guard = state.lock().await;
                     if !sweep_address_txs.is_empty() {
+                        let mut state_guard = state.lock().await;
                         state_guard.success_sweep(id, sweep_address_txs).await;
-                    } else if !sweep_address_task.retry {
-                        state_guard.fail_sweep(id).await;
+                        drop(state_guard);
                     }
-                    drop(state_guard);
                 }
             }
             tokio::time::sleep(std::time::Duration::from_secs(5)).await;
diff --git a/src/syncerd/syncer_state.rs b/src/syncerd/syncer_state.rs
index 7a4c50e9..121d36e1 100644
--- a/src/syncerd/syncer_state.rs
+++ b/src/syncerd/syncer_state.rs
@@ -642,36 +642,6 @@ impl SyncerState {
         self.drop_lifetimes();
     }
 
-    pub async fn fail_sweep(&mut self, id: &InternalId) {
-        if let Some(sweep_address) = self.sweep_addresses.get(id) {
-            send_event(
-                &self.tx_event,
-                &mut vec![(
-                    Event::TaskAborted(TaskAborted {
-                        id: vec![],
-                        error: Some(
-                            "Sweep failed, did not find any assets associated with the address"
-                                .to_string(),
-                        ),
-                    }),
-                    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 {
@@ -1063,7 +1033,6 @@ async fn syncer_state_sweep_addresses() {
     let sweep_task = SweepAddress {
         id: TaskId(0),
         lifetime: 11,
-        retry: true,
         from_height: None,
         addendum: SweepAddressAddendum::Monero(SweepXmrAddress {
             view_key: monero::PrivateKey::from_str(
diff --git a/src/syncerd/types.rs b/src/syncerd/types.rs
index f5f8a345..4c8ee9ed 100644
--- a/src/syncerd/types.rs
+++ b/src/syncerd/types.rs
@@ -74,7 +74,6 @@ impl StrictDecode for XmrAddressAddendum {
 #[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
 #[display(Debug)]
 pub struct SweepAddress {
-    pub retry: bool,
     pub id: TaskId,
     pub lifetime: u64,
     pub addendum: SweepAddressAddendum,


Test patch:
diff --git a/tests/functional.rs b/tests/functional.rs
index 14432457..4fe93b9d 100644
--- a/tests/functional.rs
+++ b/tests/functional.rs
@@ -977,7 +977,6 @@ fn bitcoin_syncer_sweep_address_test() {
             id: TaskId(0),
             lifetime: blocks,
             from_height: None,
-            retry: true,
             addendum: SweepAddressAddendum::Bitcoin(SweepBitcoinAddress {
                 source_private_key: (&wif_private_key.to_bytes()[..]).try_into().unwrap(),
                 source_address: sweep_source_address.clone(),
@@ -1039,7 +1038,6 @@ fn bitcoin_syncer_sweep_address_test() {
             id: TaskId(0),
             lifetime: blocks,
             from_height: None,
-            retry: true,
             addendum: SweepAddressAddendum::Bitcoin(SweepBitcoinAddress {
                 source_private_key: (&wif_private_key.to_bytes()[..]).try_into().unwrap(),
                 source_address: sweep_source_address,
@@ -1250,7 +1248,6 @@ async fn monero_syncer_sweep_test() {
             id: TaskId(0),
             lifetime: blocks + 40,
             from_height: None,
-            retry: true,
             addendum: SweepAddressAddendum::Monero(SweepXmrAddress {
                 spend_key,
                 view_key,