Use bmrng to model communicaton of Bob's EventLoop with the handle

bmrng is a library providing a request-response channel that allows
the receiving end of the channel to send a response back to the sender.
This allows us to more accurately implement the functions on the
`EventLoopHandle`. In particular, we now _wait_ for the ACK of specific
messages from the other party before resolving the future.

For example, when sending the encrypted signature, the async function
on the `EventLoopHandle` does not resolve until we received the ACK
from the other party.

We also delete the `Channels` abstraction in favor of directly creating
bmrng channels. This allows us to directly control the channel buffer
which we set to 1 because we don't need more than that on Bob's side.

Cargo.lock

@@ -481,6 +481,17 @@ dependencies = [
  "byte-tools",
 ]
 
+[[package]]
+name = "bmrng"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2ba7ad917af2fc43efa0b20d2bf3b2c1bd1090fa2a6b8c73847458c8335dea2b"
+dependencies = [
+ "futures",
+ "loom",
+ "tokio 1.4.0",
+]
+
 [[package]]
 name = "bs58"
 version = "0.4.0"
@@ -1181,6 +1192,19 @@ dependencies = [
  "byteorder",
 ]
 
+[[package]]
+name = "generator"
+version = "0.6.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "061d3be1afec479d56fa3bd182bf966c7999ec175fcfdb87ac14d417241366c6"
+dependencies = [
+ "cc",
+ "libc",
+ "log 0.4.14",
+ "rustversion",
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "generic-array"
 version = "0.12.4"
@@ -1909,6 +1933,20 @@ dependencies = [
  "cfg-if 1.0.0",
 ]
 
+[[package]]
+name = "loom"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d44c73b4636e497b4917eb21c33539efa3816741a2d3ff26c6316f1b529481a4"
+dependencies = [
+ "cfg-if 1.0.0",
+ "futures-util",
+ "generator",
+ "scoped-tls",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "lru"
 version = "0.6.5"
@@ -3014,6 +3052,12 @@ dependencies = [
  "webpki",
 ]
 
+[[package]]
+name = "rustversion"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb5d2a036dc6d2d8fd16fde3498b04306e29bd193bf306a57427019b823d5acd"
+
 [[package]]
 name = "rw-stream-sink"
 version = "0.2.1"
@@ -3037,6 +3081,12 @@ version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ef703b7cb59335eae2eb93ceb664c0eb7ea6bf567079d843e09420219668e072"
 
+[[package]]
+name = "scoped-tls"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ea6a9290e3c9cf0f18145ef7ffa62d68ee0bf5fcd651017e586dc7fd5da448c2"
+
 [[package]]
 name = "scopeguard"
 version = "1.1.0"
@@ -3522,6 +3572,7 @@ dependencies = [
  "big-bytes",
  "bitcoin",
  "bitcoin-harness",
+ "bmrng",
  "config",
  "conquer-once",
  "curve25519-dalek",

swap/Cargo.toml

@@ -18,6 +18,7 @@ base64 = "0.13"
 bdk = { version = "0.5" }
 big-bytes = "1"
 bitcoin = { version = "0.26", features = ["rand", "use-serde"] }
+bmrng = "0.5"
 config = { version = "0.11", default-features = false, features = ["toml"] }
 conquer-once = "0.3"
 curve25519-dalek = "3"

swap/src/protocol/bob.rs

@@ -5,7 +5,9 @@ use crate::protocol::bob;
 use crate::{bitcoin, monero};
 use anyhow::{anyhow, Error, Result};
 use libp2p::core::Multiaddr;
-use libp2p::request_response::{RequestResponseEvent, RequestResponseMessage, ResponseChannel};
+use libp2p::request_response::{
+    RequestId, RequestResponseEvent, RequestResponseMessage, ResponseChannel,
+};
 use libp2p::{NetworkBehaviour, PeerId};
 use std::sync::Arc;
 use uuid::Uuid;
@@ -107,14 +109,22 @@ impl Builder {
 
 #[derive(Debug)]
 pub enum OutEvent {
-    QuoteReceived(BidQuote),
+    QuoteReceived {
-    SpotPriceReceived(spot_price::Response),
+        id: RequestId,
-    ExecutionSetupDone(Result<Box<State2>>),
+        response: BidQuote,
+    },
+    SpotPriceReceived {
+        id: RequestId,
+        response: spot_price::Response,
+    },
+    ExecutionSetupDone(Box<Result<State2>>),
     TransferProofReceived {
         msg: Box<transfer_proof::Request>,
         channel: ResponseChannel<()>,
     },
-    EncryptedSignatureAcknowledged,
+    EncryptedSignatureAcknowledged {
+        id: RequestId,
+    },
     ResponseSent, // Same variant is used for all messages as no processing is done
     CommunicationError(Error),
 }
@@ -133,7 +143,13 @@ impl From<quote::Message> for OutEvent {
     fn from(message: quote::Message) -> Self {
         match message {
             quote::Message::Request { .. } => OutEvent::unexpected_request(),
-            quote::Message::Response { response, .. } => OutEvent::QuoteReceived(response),
+            quote::Message::Response {
+                response,
+                request_id,
+            } => OutEvent::QuoteReceived {
+                id: request_id,
+                response,
+            },
         }
     }
 }
@@ -142,7 +158,13 @@ impl From<spot_price::Message> for OutEvent {
     fn from(message: spot_price::Message) -> Self {
         match message {
             spot_price::Message::Request { .. } => OutEvent::unexpected_request(),
-            spot_price::Message::Response { response, .. } => OutEvent::SpotPriceReceived(response),
+            spot_price::Message::Response {
+                response,
+                request_id,
+            } => OutEvent::SpotPriceReceived {
+                id: request_id,
+                response,
+            },
         }
     }
 }
@@ -165,8 +187,8 @@ impl From<encrypted_signature::Message> for OutEvent {
     fn from(message: encrypted_signature::Message) -> Self {
         match message {
             encrypted_signature::Message::Request { .. } => OutEvent::unexpected_request(),
-            encrypted_signature::Message::Response { .. } => {
+            encrypted_signature::Message::Response { request_id, .. } => {
-                OutEvent::EncryptedSignatureAcknowledged
+                OutEvent::EncryptedSignatureAcknowledged { id: request_id }
             }
         }
     }
@@ -221,7 +243,7 @@ where
 impl From<execution_setup::OutEvent> for OutEvent {
     fn from(event: execution_setup::OutEvent) -> Self {
         match event {
-            execution_setup::OutEvent::Done(res) => OutEvent::ExecutionSetupDone(res.map(Box::new)),
+            execution_setup::OutEvent::Done(res) => OutEvent::ExecutionSetupDone(Box::new(res)),
         }
     }
 }

swap/src/protocol/bob/event_loop.rs

@@ -3,27 +3,46 @@ use crate::network::quote::BidQuote;
 use crate::network::{encrypted_signature, spot_price, transfer_proof};
 use crate::protocol::bob::{Behaviour, OutEvent, State0, State2};
 use crate::{bitcoin, monero};
-use anyhow::{anyhow, Result};
+use anyhow::{Context, Result};
-use futures::FutureExt;
+use futures::future::{BoxFuture, OptionFuture};
+use futures::{FutureExt, StreamExt};
+use libp2p::request_response::{RequestId, ResponseChannel};
 use libp2p::swarm::SwarmEvent;
 use libp2p::{PeerId, Swarm};
+use std::collections::HashMap;
 use std::sync::Arc;
-use tokio::sync::mpsc::{Receiver, Sender};
+use std::time::Duration;
-use tracing::{debug, error};
 
 #[allow(missing_debug_implementations)]
 pub struct EventLoop {
     swarm: libp2p::Swarm<Behaviour>,
     bitcoin_wallet: Arc<bitcoin::Wallet>,
     alice_peer_id: PeerId,
-    request_spot_price: Receiver<spot_price::Request>,
+
-    recv_spot_price: Sender<spot_price::Response>,
+    // these streams represents outgoing requests that we have to make
-    start_execution_setup: Receiver<State0>,
+    quote_requests: bmrng::RequestReceiverStream<(), BidQuote>,
-    done_execution_setup: Sender<Result<State2>>,
+    spot_price_requests: bmrng::RequestReceiverStream<spot_price::Request, spot_price::Response>,
-    recv_transfer_proof: Sender<transfer_proof::Request>,
+    encrypted_signature_requests: bmrng::RequestReceiverStream<encrypted_signature::Request, ()>,
-    send_encrypted_signature: Receiver<encrypted_signature::Request>,
+    execution_setup_requests: bmrng::RequestReceiverStream<State0, Result<State2>>,
-    request_quote: Receiver<()>,
+
-    recv_quote: Sender<BidQuote>,
+    // these represents requests that are currently in-flight.
+    // once we get a response to a matching [`RequestId`], we will use the responder to relay the
+    // response.
+    inflight_spot_price_requests: HashMap<RequestId, bmrng::Responder<spot_price::Response>>,
+    inflight_quote_requests: HashMap<RequestId, bmrng::Responder<BidQuote>>,
+    inflight_encrypted_signature_requests: HashMap<RequestId, bmrng::Responder<()>>,
+    inflight_execution_setup: Option<bmrng::Responder<Result<State2>>>,
+
+    /// The sender we will use to relay incoming transfer proofs.
+    transfer_proof: bmrng::RequestSender<transfer_proof::Request, ()>,
+    /// The future representing the successful handling of an incoming transfer
+    /// proof.
+    ///
+    /// Once we've sent a transfer proof to the ongoing swap, this future waits
+    /// until the swap took it "out" of the `EventLoopHandle`. As this future
+    /// resolves, we use the `ResponseChannel` returned from it to send an ACK
+    /// to Alice that we have successfully processed the transfer proof.
+    pending_transfer_proof: OptionFuture<BoxFuture<'static, ResponseChannel<()>>>,
 }
 
 impl EventLoop {
@@ -32,38 +51,34 @@ impl EventLoop {
         alice_peer_id: PeerId,
         bitcoin_wallet: Arc<bitcoin::Wallet>,
     ) -> Result<(Self, EventLoopHandle)> {
-        let start_execution_setup = Channels::new();
+        let execution_setup = bmrng::channel_with_timeout(1, Duration::from_secs(30));
-        let done_execution_setup = Channels::new();
+        let transfer_proof = bmrng::channel_with_timeout(1, Duration::from_secs(30));
-        let recv_transfer_proof = Channels::new();
+        let encrypted_signature = bmrng::channel_with_timeout(1, Duration::from_secs(30));
-        let send_encrypted_signature = Channels::new();
+        let spot_price = bmrng::channel_with_timeout(1, Duration::from_secs(30));
-        let request_spot_price = Channels::new();
+        let quote = bmrng::channel_with_timeout(1, Duration::from_secs(30));
-        let recv_spot_price = Channels::new();
-        let request_quote = Channels::new();
-        let recv_quote = Channels::new();
 
         let event_loop = EventLoop {
             swarm,
             alice_peer_id,
             bitcoin_wallet,
-            start_execution_setup: start_execution_setup.receiver,
+            execution_setup_requests: execution_setup.1.into(),
-            done_execution_setup: done_execution_setup.sender,
+            transfer_proof: transfer_proof.0,
-            recv_transfer_proof: recv_transfer_proof.sender,
+            encrypted_signature_requests: encrypted_signature.1.into(),
-            send_encrypted_signature: send_encrypted_signature.receiver,
+            spot_price_requests: spot_price.1.into(),
-            request_spot_price: request_spot_price.receiver,
+            quote_requests: quote.1.into(),
-            recv_spot_price: recv_spot_price.sender,
+            inflight_spot_price_requests: HashMap::default(),
-            request_quote: request_quote.receiver,
+            inflight_quote_requests: HashMap::default(),
-            recv_quote: recv_quote.sender,
+            inflight_execution_setup: None,
+            inflight_encrypted_signature_requests: HashMap::default(),
+            pending_transfer_proof: OptionFuture::from(None),
         };
 
         let handle = EventLoopHandle {
-            start_execution_setup: start_execution_setup.sender,
+            execution_setup: execution_setup.0,
-            done_execution_setup: done_execution_setup.receiver,
+            transfer_proof: transfer_proof.1,
-            recv_transfer_proof: recv_transfer_proof.receiver,
+            encrypted_signature: encrypted_signature.0,
-            send_encrypted_signature: send_encrypted_signature.sender,
+            spot_price: spot_price.0,
-            request_spot_price: request_spot_price.sender,
+            quote: quote.0,
-            recv_spot_price: recv_spot_price.receiver,
-            request_quote: request_quote.sender,
-            recv_quote: recv_quote.receiver,
         };
 
         Ok((event_loop, handle))
@@ -76,24 +91,40 @@ impl EventLoop {
             tokio::select! {
                 swarm_event = self.swarm.next_event().fuse() => {
                     match swarm_event {
-                        SwarmEvent::Behaviour(OutEvent::SpotPriceReceived(msg)) => {
+                        SwarmEvent::Behaviour(OutEvent::SpotPriceReceived { id, response }) => {
-                            let _ = self.recv_spot_price.send(msg).await;
+                            if let Some(responder) = self.inflight_spot_price_requests.remove(&id) {
-                        }
+                                let _ = responder.respond(response);
-                        SwarmEvent::Behaviour(OutEvent::QuoteReceived(msg)) => {
-                            let _ = self.recv_quote.send(msg).await;
-                        }
-                        SwarmEvent::Behaviour(OutEvent::ExecutionSetupDone(res)) => {
-                            let _ = self.done_execution_setup.send(res.map(|state|*state)).await;
-                        }
-                        SwarmEvent::Behaviour(OutEvent::TransferProofReceived{ msg, channel }) => {
-                            let _ = self.recv_transfer_proof.send(*msg).await;
-                            // Send back empty response so that the request/response protocol completes.
-                            if let Err(error) = self.swarm.transfer_proof.send_response(channel, ()) {
-                                error!("Failed to send Transfer Proof ack: {:?}", error);
                             }
                         }
-                        SwarmEvent::Behaviour(OutEvent::EncryptedSignatureAcknowledged) => {
+                        SwarmEvent::Behaviour(OutEvent::QuoteReceived { id, response }) => {
-                            debug!("Alice acknowledged encrypted signature");
+                            if let Some(responder) = self.inflight_quote_requests.remove(&id) {
+                                let _ = responder.respond(response);
+                            }
+                        }
+                        SwarmEvent::Behaviour(OutEvent::ExecutionSetupDone(response)) => {
+                            if let Some(responder) = self.inflight_execution_setup.take() {
+                                let _ = responder.respond(*response);
+                            }
+                        }
+                        SwarmEvent::Behaviour(OutEvent::TransferProofReceived { msg, channel }) => {
+                            let mut responder = match self.transfer_proof.send(*msg).await {
+                                Ok(responder) => responder,
+                                Err(_) => {
+                                    tracing::warn!("Failed to pass on transfer proof");
+                                    continue;
+                                }
+                            };
+
+                            self.pending_transfer_proof = OptionFuture::from(Some(async move {
+                                let _ = responder.recv().await;
+
+                                channel
+                            }.boxed()));
+                        }
+                        SwarmEvent::Behaviour(OutEvent::EncryptedSignatureAcknowledged { id }) => {
+                            if let Some(responder) = self.inflight_encrypted_signature_requests.remove(&id) {
+                                let _ = responder.respond(());
+                            }
                         }
                         SwarmEvent::Behaviour(OutEvent::ResponseSent) => {
 
@@ -133,27 +164,26 @@ impl EventLoop {
                         _ => {}
                     }
                 },
-                spot_price_request = self.request_spot_price.recv().fuse() => {
+                Some((request, responder)) = self.spot_price_requests.next().fuse() => {
-                    if let Some(request) = spot_price_request {
+                    let id = self.swarm.spot_price.send_request(&self.alice_peer_id, request);
-                        self.swarm.spot_price.send_request(&self.alice_peer_id, request);
+                    self.inflight_spot_price_requests.insert(id, responder);
-                    }
                 },
-                quote_request = self.request_quote.recv().fuse() =>  {
+                Some(((), responder)) = self.quote_requests.next().fuse() => {
-                    if quote_request.is_some() {
+                    let id = self.swarm.quote.send_request(&self.alice_peer_id, ());
-                        self.swarm.quote.send_request(&self.alice_peer_id, ());
+                    self.inflight_quote_requests.insert(id, responder);
-                    }
                 },
-                option = self.start_execution_setup.recv().fuse() => {
+                Some((request, responder)) = self.execution_setup_requests.next().fuse() => {
-                    if let Some(state0) = option {
+                    self.swarm.execution_setup.run(self.alice_peer_id, request, self.bitcoin_wallet.clone());
-                        let _ = self
+                    self.inflight_execution_setup = Some(responder);
-                            .swarm
-                            .execution_setup.run(self.alice_peer_id, state0, self.bitcoin_wallet.clone());
-                    }
                 },
-                encrypted_signature = self.send_encrypted_signature.recv().fuse() => {
+                Some((request, responder)) = self.encrypted_signature_requests.next().fuse() => {
-                    if let Some(tx_redeem_encsig) = encrypted_signature {
+                    let id = self.swarm.encrypted_signature.send_request(&self.alice_peer_id, request);
-                        self.swarm.encrypted_signature.send_request(&self.alice_peer_id, tx_redeem_encsig);
+                    self.inflight_encrypted_signature_requests.insert(id, responder);
-                    }
+                },
+                Some(response_channel) = &mut self.pending_transfer_proof => {
+                    let _ = self.swarm.transfer_proof.send_response(response_channel, ());
+
+                    self.pending_transfer_proof = OptionFuture::from(None);
                 }
             }
         }
@@ -162,87 +192,50 @@ impl EventLoop {
 
 #[derive(Debug)]
 pub struct EventLoopHandle {
-    start_execution_setup: Sender<State0>,
+    execution_setup: bmrng::RequestSender<State0, Result<State2>>,
-    done_execution_setup: Receiver<Result<State2>>,
+    transfer_proof: bmrng::RequestReceiver<transfer_proof::Request, ()>,
-    recv_transfer_proof: Receiver<transfer_proof::Request>,
+    encrypted_signature: bmrng::RequestSender<encrypted_signature::Request, ()>,
-    send_encrypted_signature: Sender<encrypted_signature::Request>,
+    spot_price: bmrng::RequestSender<spot_price::Request, spot_price::Response>,
-    request_spot_price: Sender<spot_price::Request>,
+    quote: bmrng::RequestSender<(), BidQuote>,
-    recv_spot_price: Receiver<spot_price::Response>,
-    request_quote: Sender<()>,
-    recv_quote: Receiver<BidQuote>,
 }
 
 impl EventLoopHandle {
     pub async fn execution_setup(&mut self, state0: State0) -> Result<State2> {
-        let _ = self.start_execution_setup.send(state0).await?;
+        self.execution_setup.send_receive(state0).await?
-
-        self.done_execution_setup
-            .recv()
-            .await
-            .ok_or_else(|| anyhow!("Failed to setup execution with Alice"))?
     }
 
     pub async fn recv_transfer_proof(&mut self) -> Result<transfer_proof::Request> {
-        self.recv_transfer_proof
+        let (request, responder) = self
+            .transfer_proof
             .recv()
             .await
-            .ok_or_else(|| anyhow!("Failed to receive transfer proof from Alice"))
+            .context("Failed to receive transfer proof")?;
+        responder
+            .respond(())
+            .context("Failed to acknowledge receipt of transfer proof")?;
+
+        Ok(request)
     }
 
     pub async fn request_spot_price(&mut self, btc: bitcoin::Amount) -> Result<monero::Amount> {
-        let _ = self
+        Ok(self
-            .request_spot_price
+            .spot_price
-            .send(spot_price::Request { btc })
+            .send_receive(spot_price::Request { btc })
-            .await?;
+            .await?
-
+            .xmr)
-        let response = self
-            .recv_spot_price
-            .recv()
-            .await
-            .ok_or_else(|| anyhow!("Failed to receive spot price from Alice"))?;
-
-        Ok(response.xmr)
     }
 
     pub async fn request_quote(&mut self) -> Result<BidQuote> {
-        let _ = self.request_quote.send(()).await?;
+        Ok(self.quote.send_receive(()).await?)
-
-        let quote = self
-            .recv_quote
-            .recv()
-            .await
-            .ok_or_else(|| anyhow!("Failed to receive quote from Alice"))?;
-
-        Ok(quote)
     }
 
     pub async fn send_encrypted_signature(
         &mut self,
         tx_redeem_encsig: EncryptedSignature,
     ) -> Result<()> {
-        self.send_encrypted_signature
+        Ok(self
-            .send(encrypted_signature::Request { tx_redeem_encsig })
+            .encrypted_signature
-            .await?;
+            .send_receive(encrypted_signature::Request { tx_redeem_encsig })
-
+            .await?)
-        Ok(())
-    }
-}
-
-#[derive(Debug)]
-struct Channels<T> {
-    sender: Sender<T>,
-    receiver: Receiver<T>,
-}
-
-impl<T> Channels<T> {
-    fn new() -> Channels<T> {
-        let (sender, receiver) = tokio::sync::mpsc::channel(100);
-        Channels { sender, receiver }
-    }
-}
-
-impl<T> Default for Channels<T> {
-    fn default() -> Self {
-        Self::new()
     }
 }