farcaster-swapd-syncer race-free initiation interplay: improve sequence of msgs

doc/sequencediagram.txt

@@ -36,28 +36,29 @@ t_farcasterd -> t_farcasterd : launch\npeerd connect
 t_wallet <- t_farcasterd : Ctl TakeOffer
 t_wallet -> t_wallet : create taker wallet
 t_wallet -> t_farcasterd : Ctl LaunchSwap
-t_swap -> t_farcasterd : Ctl Hello
-t_swap -> t_wallet : Ctl Hello
 t_farcasterd -> t_farcasterd:launch syncer
+t_syncer->t_farcasterd:Ctl Hello [from Syncer Bitcoin [if newly launched] OR Swapd (if syncer reused)]
 t_swap <- t_farcasterd : Ctl TakeSwap
+t_syncer<-t_swap:Ctl EstimateFee (btc)
+t_syncer <- t_swap : Ctl WatchHeight
 t_swap -> peerd : Msg TakerCommit
 peerd -> m_farcasterd : Msg TakerCommit
-m_farcasterd -> m_wallet : Ctl BitcoinAddress
+
+m_farcasterd -> m_wallet : Msg BitcoinAddress
+m_farcasterd->m_wallet : Msg MoneroAddress
 m_farcasterd -> m_wallet : Msg TakerCommit
 m_wallet -> m_wallet : create maker wallet
 m_wallet -> m_farcasterd : Ctl LaunchSwap
-m_swap -> m_farcasterd : Ctl Hello
-m_swap -> m_wallet : Ctl Hello
 m_farcasterd -> m_farcasterd:launch syncer
 m_farcasterd -> m_swap : Ctl MakeSwap
 
-m_swap->m_syncer:If Bob, Ctl EstimateFee (btc)
+m_swap->m_syncer:Ctl EstimateFee (btc)
+m_swap->m_syncer:Ctl WatchHeight
 m_swap -> peerd : Msg MakerCommit
 t_swap <- peerd : Msg MakerCommit
 // TODO: verify that swapd launches no matter what
-t_syncer <- t_swap : Ctl WatchHeight
+m_farcasterd<-m_syncer:Ctl Hello [from Syncer Bitcoin (if newly launched) OR Swapd (if syncer reused)]
 t_syncer <- t_swap : if Bob, Watch Arbitrating Funding Address
-t_syncer<-t_swap:If Bob, Ctl EstimateFee (btc)
 t_swap -> t_wallet : Msg MakerCommit
 t_wallet -> t_swap : Ctl RevealProof (taker is sender)
 t_swap -> peerd : Msg RevealProof (taker is sender)
@@ -70,9 +71,9 @@ m_swap -> m_farcasterd : if Bob, ask for funding
 m_swap -> m_swap : if Bob, ADD PENDING Msg Reveal
 m_swap -> m_wallet : if Alice, Msg Reveal (taker is sender)
 
-m_swap->m_syncer:Ctl WatchHeight
 m_swap -> m_syncer:if Bob, Watch Arbitrating Funding Address
 m_swap <- m_syncer:If Bob, Arbitrating Funding event
+m_farcasterd<-m_swap:Ctl FundingCompleted Bitcoin
 m_swap->m_wallet:if Bob, Ctl Tx::Funding
 m_swap<-m_wallet:If Bob, Ctl FundingUpdated
 m_swap -> m_wallet : if Bob, SEND PENDING Msg RevealProof (taker is sender) 
@@ -88,6 +89,7 @@ t_swap -> t_farcasterd : if Bob, ask for funding
 t_swap -> t_swap : if Bob, ADD PENDING Msg Reveal
 t_swap -> t_wallet : if Alice, Msg Reveal (maker is sender)
 t_syncer -> t_swap:If Bob, Arbitrating Funding event
+t_swap->t_farcasterd:Ctl FundingCompleted Bitcoin
 t_swap->t_wallet:if Bob, Ctl Tx::Funding
 t_swap<-t_wallet:If Bob, Ctl FundingUpdated
 t_swap -> t_wallet : if Bob, SEND PENDING Msg RevealProof (maker is sender)

src/farcasterd/runtime.rs

@@ -135,6 +135,7 @@ pub fn run(
         spawning_services: none!(),
         making_swaps: none!(),
         taking_swaps: none!(),
+        pending_swap_init: none!(),
         arb_addrs: none!(),
         acc_addrs: none!(),
         public_offers: none!(),
@@ -172,6 +173,7 @@ pub struct Runtime {
     spawning_services: HashMap<ServiceId, ServiceId>,
     making_swaps: HashMap<ServiceId, (request::InitSwap, Network)>,
     taking_swaps: HashMap<ServiceId, (request::InitSwap, Network)>,
+    pending_swap_init: HashMap<ServiceId, Vec<(Request, SwapId)>>,
     public_offers: HashSet<PublicOffer<BtcXmr>>,
     arb_addrs: HashMap<PublicOfferId, bitcoin::Address>,
     acc_addrs: HashMap<PublicOfferId, monero::Address>,
@@ -616,6 +618,21 @@ impl Runtime {
                                 request,
                             )?;
                         }
+                        let xs = self.pending_swap_init.get(&source).cloned();
+                        if let Some(xs) = xs {
+                            for (init_swap_req, swap_id) in xs {
+                                if let Ok(()) = endpoints.send_to(
+                                    ServiceBus::Ctl,
+                                    self.identity(),
+                                    ServiceId::Swap(swap_id.clone()),
+                                    init_swap_req,
+                                ) {
+                                    self.pending_swap_init.remove(&source);
+                                } else {
+                                    error!("Failed to dispatch init swap requests containing swap params");
+                                };
+                            }
+                        }
                     }
                     ServiceId::Syncer(..) => {
                         error!(
@@ -629,12 +646,12 @@ 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!(
                         "Swapd {} is known: we spawned it to create a swap. \
-                         Requesting swapd to be the maker of this swap",
+                             Requesting swapd to be the maker of this swap",
                         source
                     );
                     report_to.push((
@@ -644,20 +661,37 @@ impl Runtime {
                             source
                         ))),
                     ));
-                    endpoints.send_to(
-                        ServiceBus::Ctl,
-                        self.identity(),
-                        source.clone(),
-                        Request::MakeSwap(swap_params.clone()),
-                    )?;
+                    // notify this swapd about its syncers that are up and
+                    // running. if syncer not ready, then swapd will be notified
+                    // on the ServiceId::Syncer(coin, network) Hello pattern. in
+                    // sum, if syncer is up, send msg immediately else wait for
+                    // syncer to say hello, and then dispatch msg
+                    let init_swap_req = Request::MakeSwap(swap_params.clone());
+                    if self
+                        .syncer_services
+                        .contains_key(&(Coin::Bitcoin, *network))
+                    {
+                        endpoints.send_to(
+                            ServiceBus::Ctl,
+                            self.identity(),
+                            source.clone(),
+                            init_swap_req,
+                        )?;
+                    } else {
+                        let xs = self
+                            .pending_swap_init
+                            .entry(ServiceId::Syncer(Coin::Bitcoin, *network))
+                            .or_insert(vec![]);
+                        xs.push((init_swap_req, swap_params.swap_id));
+                    }
                     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!(
                         "Daemon {} is known: we spawned it to create a swap. \
-                         Requesting swapd to be the taker of this swap",
+                             Requesting swapd to be the taker of this swap",
                         source
                     );
                     report_to.push((
@@ -667,12 +701,24 @@ impl Runtime {
                             source
                         ))),
                     ));
-                    endpoints.send_to(
-                        ServiceBus::Ctl,
-                        self.identity(),
-                        source.clone(),
-                        Request::TakeSwap(swap_params.clone()),
-                    )?;
+                    let init_swap_req = Request::TakeSwap(swap_params.clone());
+                    if self
+                        .syncer_services
+                        .contains_key(&(Coin::Bitcoin, *network))
+                    {
+                        endpoints.send_to(
+                            ServiceBus::Ctl,
+                            self.identity(),
+                            source.clone(),
+                            init_swap_req,
+                        )?;
+                    } else {
+                        let xs = self
+                            .pending_swap_init
+                            .entry(ServiceId::Syncer(Coin::Bitcoin, *network))
+                            .or_insert(vec![]);
+                        xs.push((init_swap_req, swap_params.swap_id));
+                    }
                     self.running_swaps.insert(swap_params.swap_id);
                     self.taking_swaps.remove(&source);
                 } else if let Some(enquirer) = self.spawning_services.get(&source) {
@@ -786,11 +832,6 @@ impl Runtime {
                         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()

src/rpc/request.rs

@@ -305,7 +305,7 @@ fn format_keys(keys: &Keys) -> String {
 pub enum Reveal {
     AliceParameters(RevealAliceParameters<BtcXmr>),
     BobParameters(RevealBobParameters<BtcXmr>),
-    Proof(RevealProof<BtcXmr>),
+    Proof(RevealProof<BtcXmr>), // FIXME should be Msg::RevealProof(RevealProof<BtcXmr>)
 }
 
 // #[cfg_attr(feature = "serde", serde_as)]
@@ -1332,6 +1332,7 @@ impl IntoSuccessOrFailure for Result<(), crate::Error> {
     }
 }
 
+// FIXME
 impl From<(SwapId, Params)> for Reveal {
     fn from(tuple: (SwapId, Params)) -> Self {
         match tuple {
@@ -1340,9 +1341,3 @@ impl From<(SwapId, Params)> for Reveal {
         }
     }
 }
-
-impl From<(SwapId, Proof<BtcXmr>)> for Reveal {
-    fn from(tuple: (SwapId, Proof<BtcXmr>)) -> Self {
-        Reveal::Proof((tuple.0, tuple.1).into())
-    }
-}