461: Add resume-only mode for the ASB r=da-kami a=da-kami

Fixes #378 

Resume-only is a maintenance mode where no swaps are accepted but unfinished swaps are resumed.
This is achieve by ignoring incoming spot-price requests (that would lead to execution setup) in the event-loop.

- [x] Refactor `spot_price`, move Alice's decision logic into dedicated `NetworkBehaviour`
- [x] Protocol (network) level tests for the `spot_price` behaviour

Co-authored-by: Daniel Karzel <daniel@comit.network>
This commit is contained in:
bors[bot] 2021-05-07 05:13:18 +00:00 committed by GitHub
commit 095d67f946
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 1080 additions and 157 deletions

View File

@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- Resume-only mode for the ASB.
When started with `--resume-only` the ASB does not accept new, incoming swap requests but only finishes swaps that are resumed upon startup.
### Fixed
- An issue where both the ASB and the CLI point to the same default directory `xmr-btc-swap` for storing data.
@ -15,6 +20,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
If you want to access data created by a previous version you will have to rename the data folder or one of the following:
1. For the CLI you can use `--data-dir` to point to the old directory.
2. For the ASB you can change the data-dir in the config file of the ASB.
- The CLI receives proper Error messages if setting up a swap with the ASB fails.
This is a breaking change because the spot-price protocol response changed.
Expected errors scenarios that are now reported back to the CLI:
1. Balance of ASB too low
2. Buy amount sent by CLI exceeds maximum buy amount accepted by ASB
3. ASB is running in resume-only mode and does not accept incoming swap requests
## [0.5.0] - 2021-04-17

View File

@ -34,6 +34,12 @@ pub enum Command {
default_value = "0.02"
)]
ask_spread: Decimal,
#[structopt(
long = "resume-only",
help = "For maintenance only. When set, no new swap requests will be accepted, but existing unfinished swaps will be resumed."
)]
resume_only: bool,
},
History,
WithdrawBtc {

View File

@ -81,6 +81,7 @@ async fn main() -> Result<()> {
Command::Start {
max_buy,
ask_spread,
resume_only,
} => {
let bitcoin_wallet = init_bitcoin_wallet(&config, &seed, env_config).await?;
let monero_wallet = init_monero_wallet(&config, env_config).await?;
@ -118,7 +119,17 @@ async fn main() -> Result<()> {
}
};
let mut swarm = swarm::alice(&seed)?;
let current_balance = monero_wallet.get_balance().await?;
let lock_fee = monero_wallet.static_tx_fee_estimate();
let kraken_rate = KrakenRate::new(ask_spread, kraken_price_updates);
let mut swarm = swarm::alice(
&seed,
current_balance,
lock_fee,
max_buy,
kraken_rate.clone(),
resume_only,
)?;
for listen in config.network.listen {
Swarm::listen_on(&mut swarm, listen.clone())
@ -131,7 +142,7 @@ async fn main() -> Result<()> {
Arc::new(bitcoin_wallet),
Arc::new(monero_wallet),
Arc::new(db),
KrakenRate::new(ask_spread, kraken_price_updates),
kraken_rate,
max_buy,
)
.unwrap();

View File

@ -192,12 +192,6 @@ pub struct InsufficientFunds {
pub actual: Amount,
}
#[derive(Debug, Clone, Copy, thiserror::Error)]
#[error("The balance is too low, current balance: {balance}")]
pub struct BalanceTooLow {
pub balance: Amount,
}
#[derive(thiserror::Error, Debug, Clone, PartialEq)]
#[error("Overflow, cannot convert {0} to u64")]
pub struct OverflowError(pub String);

View File

@ -10,3 +10,6 @@ pub mod swarm;
pub mod tor_transport;
pub mod transfer_proof;
pub mod transport;
#[cfg(any(test, feature = "test"))]
pub mod test;

View File

@ -1,17 +1,12 @@
use crate::monero;
use crate::network::cbor_request_response::CborCodec;
use crate::protocol::{alice, bob};
use crate::{bitcoin, monero};
use libp2p::core::ProtocolName;
use libp2p::request_response::{
ProtocolSupport, RequestResponse, RequestResponseConfig, RequestResponseEvent,
RequestResponseMessage,
};
use libp2p::PeerId;
use libp2p::request_response::{RequestResponse, RequestResponseEvent, RequestResponseMessage};
use serde::{Deserialize, Serialize};
const PROTOCOL: &str = "/comit/xmr/btc/spot-price/1.0.0";
type OutEvent = RequestResponseEvent<Request, Response>;
type Message = RequestResponseMessage<Request, Response>;
pub const PROTOCOL: &str = "/comit/xmr/btc/spot-price/1.0.0";
pub type OutEvent = RequestResponseEvent<Request, Response>;
pub type Message = RequestResponseMessage<Request, Response>;
pub type Behaviour = RequestResponse<CborCodec<SpotPriceProtocol, Request, Response>>;
@ -40,62 +35,62 @@ pub struct Request {
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Response {
pub xmr: monero::Amount,
pub enum Response {
Xmr(monero::Amount),
Error(Error),
}
/// Constructs a new instance of the `spot-price` behaviour to be used by Alice.
///
/// Alice only supports inbound connections, i.e. providing spot prices for BTC
/// in XMR.
pub fn alice() -> Behaviour {
Behaviour::new(
CborCodec::default(),
vec![(SpotPriceProtocol, ProtocolSupport::Inbound)],
RequestResponseConfig::default(),
)
}
/// Constructs a new instance of the `spot-price` behaviour to be used by Bob.
///
/// Bob only supports outbound connections, i.e. requesting a spot price for a
/// given amount of BTC in XMR.
pub fn bob() -> Behaviour {
Behaviour::new(
CborCodec::default(),
vec![(SpotPriceProtocol, ProtocolSupport::Outbound)],
RequestResponseConfig::default(),
)
}
impl From<(PeerId, Message)> for alice::OutEvent {
fn from((peer, message): (PeerId, Message)) -> Self {
match message {
Message::Request {
request, channel, ..
} => Self::SpotPriceRequested {
request,
channel,
peer,
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum Error {
NoSwapsAccepted,
MaxBuyAmountExceeded {
#[serde(with = "::bitcoin::util::amount::serde::as_sat")]
max: bitcoin::Amount,
#[serde(with = "::bitcoin::util::amount::serde::as_sat")]
buy: bitcoin::Amount,
},
Message::Response { .. } => Self::unexpected_response(peer),
}
}
}
crate::impl_from_rr_event!(OutEvent, alice::OutEvent, PROTOCOL);
impl From<(PeerId, Message)> for bob::OutEvent {
fn from((peer, message): (PeerId, Message)) -> Self {
match message {
Message::Request { .. } => Self::unexpected_request(peer),
Message::Response {
response,
request_id,
} => Self::SpotPriceReceived {
id: request_id,
response,
BalanceTooLow {
#[serde(with = "::bitcoin::util::amount::serde::as_sat")]
buy: bitcoin::Amount,
},
/// To be used for errors that cannot be explained on the CLI side (e.g.
/// rate update problems on the seller side)
Other,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::monero;
#[test]
fn snapshot_test_serialize() {
let amount = monero::Amount::from_piconero(100_000u64);
let xmr = r#"{"Xmr":100000}"#.to_string();
let serialized = serde_json::to_string(&Response::Xmr(amount)).unwrap();
assert_eq!(xmr, serialized);
let error = r#"{"Error":"NoSwapsAccepted"}"#.to_string();
let serialized = serde_json::to_string(&Response::Error(Error::NoSwapsAccepted)).unwrap();
assert_eq!(error, serialized);
let error = r#"{"Error":{"MaxBuyAmountExceeded":{"max":0,"buy":0}}}"#.to_string();
let serialized = serde_json::to_string(&Response::Error(Error::MaxBuyAmountExceeded {
max: Default::default(),
buy: Default::default(),
}))
.unwrap();
assert_eq!(error, serialized);
let error = r#"{"Error":{"BalanceTooLow":{"buy":0}}}"#.to_string();
let serialized = serde_json::to_string(&Response::Error(Error::BalanceTooLow {
buy: Default::default(),
}))
.unwrap();
assert_eq!(error, serialized);
let error = r#"{"Error":"Other"}"#.to_string();
let serialized = serde_json::to_string(&Response::Error(Error::Other)).unwrap();
assert_eq!(error, serialized);
}
}
}
crate::impl_from_rr_event!(OutEvent, bob::OutEvent, PROTOCOL);

View File

@ -1,13 +1,28 @@
use crate::network::transport;
use crate::protocol::alice::event_loop::LatestRate;
use crate::protocol::{alice, bob};
use crate::seed::Seed;
use crate::tor;
use crate::{monero, tor};
use anyhow::Result;
use libp2p::swarm::{NetworkBehaviour, SwarmBuilder};
use libp2p::{PeerId, Swarm};
use std::fmt::Debug;
pub fn alice(seed: &Seed) -> Result<Swarm<alice::Behaviour>> {
with_clear_net(seed, alice::Behaviour::default())
pub fn alice<LR>(
seed: &Seed,
balance: monero::Amount,
lock_fee: monero::Amount,
max_buy: bitcoin::Amount,
latest_rate: LR,
resume_only: bool,
) -> Result<Swarm<alice::Behaviour<LR>>>
where
LR: LatestRate + Send + 'static + Debug,
{
with_clear_net(
seed,
alice::Behaviour::new(balance, lock_fee, max_buy, latest_rate, resume_only),
)
}
pub async fn bob(

162
swap/src/network/test.rs Normal file
View File

@ -0,0 +1,162 @@
use futures::future;
use libp2p::core::muxing::StreamMuxerBox;
use libp2p::core::transport::memory::MemoryTransport;
use libp2p::core::upgrade::{SelectUpgrade, Version};
use libp2p::core::{Executor, Multiaddr};
use libp2p::mplex::MplexConfig;
use libp2p::noise::{self, NoiseConfig, X25519Spec};
use libp2p::swarm::{
IntoProtocolsHandler, NetworkBehaviour, ProtocolsHandler, SwarmBuilder, SwarmEvent,
};
use libp2p::{identity, yamux, PeerId, Swarm, Transport};
use std::fmt::Debug;
use std::future::Future;
use std::pin::Pin;
use std::time::Duration;
use tokio::time;
/// An adaptor struct for libp2p that spawns futures into the current
/// thread-local runtime.
struct GlobalSpawnTokioExecutor;
impl Executor for GlobalSpawnTokioExecutor {
fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
let _ = tokio::spawn(future);
}
}
#[allow(missing_debug_implementations)]
pub struct Actor<B: NetworkBehaviour> {
pub swarm: Swarm<B>,
pub addr: Multiaddr,
pub peer_id: PeerId,
}
pub async fn new_connected_swarm_pair<B, F>(behaviour_fn: F) -> (Actor<B>, Actor<B>)
where
B: NetworkBehaviour,
F: Fn(PeerId, identity::Keypair) -> B + Clone,
<<<B as NetworkBehaviour>::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::InEvent: Clone,
<B as NetworkBehaviour>::OutEvent: Debug{
let (swarm, addr, peer_id) = new_swarm(behaviour_fn.clone());
let mut alice = Actor {
swarm,
addr,
peer_id,
};
let (swarm, addr, peer_id) = new_swarm(behaviour_fn);
let mut bob = Actor {
swarm,
addr,
peer_id,
};
connect(&mut alice.swarm, &mut bob.swarm).await;
(alice, bob)
}
pub fn new_swarm<B: NetworkBehaviour, F: Fn(PeerId, identity::Keypair) -> B>(
behaviour_fn: F,
) -> (Swarm<B>, Multiaddr, PeerId)
where
B: NetworkBehaviour,
{
let id_keys = identity::Keypair::generate_ed25519();
let peer_id = PeerId::from(id_keys.public());
let dh_keys = noise::Keypair::<X25519Spec>::new()
.into_authentic(&id_keys)
.expect("failed to create dh_keys");
let noise = NoiseConfig::xx(dh_keys).into_authenticated();
let transport = MemoryTransport::default()
.upgrade(Version::V1)
.authenticate(noise)
.multiplex(SelectUpgrade::new(
yamux::YamuxConfig::default(),
MplexConfig::new(),
))
.timeout(Duration::from_secs(5))
.map(|(peer, muxer), _| (peer, StreamMuxerBox::new(muxer)))
.boxed();
let mut swarm: Swarm<B> = SwarmBuilder::new(transport, behaviour_fn(peer_id, id_keys), peer_id)
.executor(Box::new(GlobalSpawnTokioExecutor))
.build();
let address_port = rand::random::<u64>();
let addr = format!("/memory/{}", address_port)
.parse::<Multiaddr>()
.unwrap();
Swarm::listen_on(&mut swarm, addr.clone()).unwrap();
(swarm, addr, peer_id)
}
pub async fn await_events_or_timeout<A, B>(
alice_event: impl Future<Output = A>,
bob_event: impl Future<Output = B>,
) -> (A, B) {
time::timeout(
Duration::from_secs(10),
future::join(alice_event, bob_event),
)
.await
.expect("network behaviours to emit an event within 10 seconds")
}
/// Connects two swarms with each other.
///
/// This assumes the transport that is in use can be used by Bob to connect to
/// the listen address that is emitted by Alice. In other words, they have to be
/// on the same network. The memory transport used by the above `new_swarm`
/// function fulfills this.
///
/// We also assume that the swarms don't emit any behaviour events during the
/// connection phase. Any event emitted is considered a bug from this functions
/// PoV because they would be lost.
pub async fn connect<BA, BB>(alice: &mut Swarm<BA>, bob: &mut Swarm<BB>)
where
BA: NetworkBehaviour,
BB: NetworkBehaviour,
<BA as NetworkBehaviour>::OutEvent: Debug,
<BB as NetworkBehaviour>::OutEvent: Debug,
{
let mut alice_connected = false;
let mut bob_connected = false;
while !alice_connected && !bob_connected {
let (alice_event, bob_event) = future::join(alice.next_event(), bob.next_event()).await;
match alice_event {
SwarmEvent::ConnectionEstablished { .. } => {
alice_connected = true;
}
SwarmEvent::NewListenAddr(addr) => {
bob.dial_addr(addr).unwrap();
}
SwarmEvent::Behaviour(event) => {
panic!(
"alice unexpectedly emitted a behaviour event during connection: {:?}",
event
);
}
_ => {}
}
match bob_event {
SwarmEvent::ConnectionEstablished { .. } => {
bob_connected = true;
}
SwarmEvent::Behaviour(event) => {
panic!(
"bob unexpectedly emitted a behaviour event during connection: {:?}",
event
);
}
_ => {}
}
}
}

View File

@ -14,6 +14,7 @@ pub use self::swap::{run, run_until};
mod behaviour;
pub mod event_loop;
mod execution_setup;
mod spot_price;
pub mod state;
pub mod swap;

View File

@ -1,6 +1,8 @@
use crate::monero;
use crate::network::quote::BidQuote;
use crate::network::{encrypted_signature, quote, spot_price, transfer_proof};
use crate::protocol::alice::{execution_setup, State3};
use crate::network::{encrypted_signature, quote, transfer_proof};
use crate::protocol::alice::event_loop::LatestRate;
use crate::protocol::alice::{execution_setup, spot_price, State3};
use anyhow::{anyhow, Error};
use libp2p::request_response::{RequestId, ResponseChannel};
use libp2p::{NetworkBehaviour, PeerId};
@ -8,10 +10,14 @@ use uuid::Uuid;
#[derive(Debug)]
pub enum OutEvent {
SpotPriceRequested {
request: spot_price::Request,
channel: ResponseChannel<spot_price::Response>,
SwapRequestDeclined {
peer: PeerId,
error: spot_price::Error,
},
ExecutionSetupStart {
peer: PeerId,
btc: bitcoin::Amount,
xmr: monero::Amount,
},
QuoteRequested {
channel: ResponseChannel<BidQuote>,
@ -60,19 +66,37 @@ impl OutEvent {
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "OutEvent", event_process = false)]
#[allow(missing_debug_implementations)]
pub struct Behaviour {
pub struct Behaviour<LR>
where
LR: LatestRate + Send + 'static,
{
pub quote: quote::Behaviour,
pub spot_price: spot_price::Behaviour,
pub spot_price: spot_price::Behaviour<LR>,
pub execution_setup: execution_setup::Behaviour,
pub transfer_proof: transfer_proof::Behaviour,
pub encrypted_signature: encrypted_signature::Behaviour,
}
impl Default for Behaviour {
fn default() -> Self {
impl<LR> Behaviour<LR>
where
LR: LatestRate + Send + 'static,
{
pub fn new(
balance: monero::Amount,
lock_fee: monero::Amount,
max_buy: bitcoin::Amount,
latest_rate: LR,
resume_only: bool,
) -> Self {
Self {
quote: quote::alice(),
spot_price: spot_price::alice(),
spot_price: spot_price::Behaviour::new(
balance,
lock_fee,
max_buy,
latest_rate,
resume_only,
),
execution_setup: Default::default(),
transfer_proof: transfer_proof::alice(),
encrypted_signature: encrypted_signature::alice(),

View File

@ -1,12 +1,12 @@
use crate::asb::Rate;
use crate::database::Database;
use crate::env::Config;
use crate::monero::BalanceTooLow;
use crate::network::quote::BidQuote;
use crate::network::{spot_price, transfer_proof};
use crate::network::transfer_proof;
use crate::protocol::alice::spot_price::Error;
use crate::protocol::alice::{AliceState, Behaviour, OutEvent, State0, State3, Swap};
use crate::{bitcoin, kraken, monero};
use anyhow::{bail, Context, Result};
use anyhow::{Context, Result};
use futures::future;
use futures::future::{BoxFuture, FutureExt};
use futures::stream::{FuturesUnordered, StreamExt};
@ -17,6 +17,7 @@ use rand::rngs::OsRng;
use rust_decimal::Decimal;
use std::collections::HashMap;
use std::convert::Infallible;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
use uuid::Uuid;
@ -32,13 +33,16 @@ type OutgoingTransferProof =
BoxFuture<'static, Result<(PeerId, transfer_proof::Request, bmrng::Responder<()>)>>;
#[allow(missing_debug_implementations)]
pub struct EventLoop<RS> {
swarm: libp2p::Swarm<Behaviour>,
pub struct EventLoop<LR>
where
LR: LatestRate + Send + 'static + Debug,
{
swarm: libp2p::Swarm<Behaviour<LR>>,
env_config: Config,
bitcoin_wallet: Arc<bitcoin::Wallet>,
monero_wallet: Arc<monero::Wallet>,
db: Arc<Database>,
latest_rate: RS,
latest_rate: LR,
max_buy: bitcoin::Amount,
swap_sender: mpsc::Sender<Swap>,
@ -60,10 +64,11 @@ pub struct EventLoop<RS> {
impl<LR> EventLoop<LR>
where
LR: LatestRate,
LR: LatestRate + Send + 'static + Debug,
{
#[allow(clippy::too_many_arguments)]
pub fn new(
swarm: Swarm<Behaviour>,
swarm: Swarm<Behaviour<LR>>,
env_config: Config,
bitcoin_wallet: Arc<bitcoin::Wallet>,
monero_wallet: Arc<monero::Wallet>,
@ -143,24 +148,8 @@ where
tokio::select! {
swarm_event = self.swarm.next_event() => {
match swarm_event {
SwarmEvent::Behaviour(OutEvent::SpotPriceRequested { request, channel, peer }) => {
let btc = request.btc;
let xmr = match self.handle_spot_price_request(btc, self.monero_wallet.clone()).await {
Ok(xmr) => xmr,
Err(e) => {
tracing::warn!(%peer, "Failed to produce spot price for {}: {:#}", btc, e);
continue;
}
};
SwarmEvent::Behaviour(OutEvent::ExecutionSetupStart { peer, btc, xmr }) => {
match self.swarm.behaviour_mut().spot_price.send_response(channel, spot_price::Response { xmr }) {
Ok(_) => {},
Err(_) => {
// if we can't respond, the peer probably just disconnected so it is not a huge deal, only log this on debug
tracing::debug!(%peer, "Failed to respond with spot price");
continue;
}
}
let tx_redeem_fee = self.bitcoin_wallet
.estimate_fee(bitcoin::TxRedeem::weight(), btc)
.await;
@ -178,7 +167,7 @@ where
(redeem_address, punish_address)
}
_ => {
tracing::error!("Could not get new address.");
tracing::error!(%peer, "Failed to get new address during execution setup.");
continue;
}
};
@ -191,7 +180,7 @@ where
(tx_redeem_fee, tx_punish_fee)
}
_ => {
tracing::error!("Could not calculate transaction fees.");
tracing::error!(%peer, "Failed to calculate transaction fees during execution setup.");
continue;
}
};
@ -215,7 +204,30 @@ where
self.swarm.behaviour_mut().execution_setup.run(peer, state0);
}
SwarmEvent::Behaviour(OutEvent::SwapRequestDeclined { peer, error }) => {
match error {
Error::ResumeOnlyMode | Error::MaxBuyAmountExceeded { .. } => {
tracing::warn!(%peer, "Ignoring spot price request because: {}", error);
}
Error::BalanceTooLow { .. }
| Error::LatestRateFetchFailed(_)
| Error::SellQuoteCalculationFailed(_) => {
tracing::error!(%peer, "Ignoring spot price request because: {}", error);
}
}
}
SwarmEvent::Behaviour(OutEvent::QuoteRequested { channel, peer }) => {
// TODO: Move the spot-price update into dedicated update stream to decouple it from quote requests
let current_balance = self.monero_wallet.get_balance().await;
match current_balance {
Ok(balance) => {
self.swarm.behaviour_mut().spot_price.update_balance(balance);
}
Err(e) => {
tracing::error!("Failed to fetch Monero balance: {:#}", e);
}
}
let quote = match self.make_quote(self.max_buy).await {
Ok(quote) => quote,
Err(e) => {
@ -343,36 +355,6 @@ where
}
}
async fn handle_spot_price_request(
&mut self,
btc: bitcoin::Amount,
monero_wallet: Arc<monero::Wallet>,
) -> Result<monero::Amount> {
let rate = self
.latest_rate
.latest_rate()
.context("Failed to get latest rate")?;
if btc > self.max_buy {
bail!(MaximumBuyAmountExceeded {
actual: btc,
max: self.max_buy
})
}
let xmr_balance = monero_wallet.get_balance().await?;
let xmr_lock_fees = monero_wallet.static_tx_fee_estimate();
let xmr = rate.sell_quote(btc)?;
if xmr_balance < xmr + xmr_lock_fees {
bail!(BalanceTooLow {
balance: xmr_balance
})
}
Ok(xmr)
}
async fn make_quote(&mut self, max_buy: bitcoin::Amount) -> Result<BidQuote> {
let rate = self
.latest_rate
@ -491,7 +473,7 @@ impl LatestRate for FixedRate {
/// Produces [`Rate`]s based on [`PriceUpdate`]s from kraken and a configured
/// spread.
#[derive(Debug)]
#[derive(Debug, Clone)]
pub struct KrakenRate {
ask_spread: Decimal,
price_updates: kraken::PriceUpdates,
@ -551,13 +533,6 @@ impl EventLoopHandle {
}
}
#[derive(Debug, Clone, Copy, thiserror::Error)]
#[error("Refusing to buy {actual} because the maximum configured limit is {max}")]
pub struct MaximumBuyAmountExceeded {
pub max: bitcoin::Amount,
pub actual: bitcoin::Amount,
}
#[allow(missing_debug_implementations)]
struct MpscChannels<T> {
sender: mpsc::Sender<T>,

View File

@ -0,0 +1,630 @@
use crate::monero;
use crate::network::cbor_request_response::CborCodec;
use crate::network::spot_price;
use crate::network::spot_price::SpotPriceProtocol;
use crate::protocol::alice;
use crate::protocol::alice::event_loop::LatestRate;
use libp2p::request_response::{
ProtocolSupport, RequestResponseConfig, RequestResponseEvent, RequestResponseMessage,
ResponseChannel,
};
use libp2p::swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters};
use libp2p::{NetworkBehaviour, PeerId};
use std::collections::VecDeque;
use std::fmt::Debug;
use std::task::{Context, Poll};
#[derive(Debug)]
pub enum OutEvent {
ExecutionSetupParams {
peer: PeerId,
btc: bitcoin::Amount,
xmr: monero::Amount,
},
Error {
peer: PeerId,
error: Error,
},
}
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "OutEvent", poll_method = "poll", event_process = true)]
#[allow(missing_debug_implementations)]
pub struct Behaviour<LR>
where
LR: LatestRate + Send + 'static,
{
behaviour: spot_price::Behaviour,
#[behaviour(ignore)]
events: VecDeque<OutEvent>,
#[behaviour(ignore)]
balance: monero::Amount,
#[behaviour(ignore)]
lock_fee: monero::Amount,
#[behaviour(ignore)]
max_buy: bitcoin::Amount,
#[behaviour(ignore)]
latest_rate: LR,
#[behaviour(ignore)]
resume_only: bool,
}
/// Behaviour that handles spot prices.
/// All the logic how to react to a spot price request is contained here, events
/// reporting the successful handling of a spot price request or a failure are
/// bubbled up to the parent behaviour.
impl<LR> Behaviour<LR>
where
LR: LatestRate + Send + 'static,
{
pub fn new(
balance: monero::Amount,
lock_fee: monero::Amount,
max_buy: bitcoin::Amount,
latest_rate: LR,
resume_only: bool,
) -> Self {
Self {
behaviour: spot_price::Behaviour::new(
CborCodec::default(),
vec![(SpotPriceProtocol, ProtocolSupport::Inbound)],
RequestResponseConfig::default(),
),
events: Default::default(),
balance,
lock_fee,
max_buy,
latest_rate,
resume_only,
}
}
pub fn update_balance(&mut self, balance: monero::Amount) {
self.balance = balance;
}
fn decline(
&mut self,
peer: PeerId,
channel: ResponseChannel<spot_price::Response>,
error: Error,
) {
if self
.behaviour
.send_response(
channel,
spot_price::Response::Error(error.to_error_response()),
)
.is_err()
{
tracing::debug!(%peer, "Unable to send error response for spot price request");
}
self.events.push_back(OutEvent::Error { peer, error });
}
fn poll<BIE>(
&mut self,
_cx: &mut Context<'_>,
_params: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<BIE, OutEvent>> {
if let Some(event) = self.events.pop_front() {
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
}
// We trust in libp2p to poll us.
Poll::Pending
}
}
impl<LR> NetworkBehaviourEventProcess<spot_price::OutEvent> for Behaviour<LR>
where
LR: LatestRate + Send + 'static,
{
fn inject_event(&mut self, event: spot_price::OutEvent) {
let (peer, message) = match event {
RequestResponseEvent::Message { peer, message } => (peer, message),
RequestResponseEvent::OutboundFailure { peer, error, .. } => {
tracing::error!(%peer, "Failure sending spot price response: {:#}", error);
return;
}
RequestResponseEvent::InboundFailure { peer, error, .. } => {
tracing::warn!(%peer, "Inbound failure when handling spot price request: {:#}", error);
return;
}
RequestResponseEvent::ResponseSent { peer, .. } => {
tracing::debug!(%peer, "Spot price response sent");
return;
}
};
let (request, channel) = match message {
RequestResponseMessage::Request {
request, channel, ..
} => (request, channel),
RequestResponseMessage::Response { .. } => {
tracing::error!("Unexpected message");
return;
}
};
if self.resume_only {
self.decline(peer, channel, Error::ResumeOnlyMode);
return;
}
let btc = request.btc;
if btc > self.max_buy {
self.decline(peer, channel, Error::MaxBuyAmountExceeded {
max: self.max_buy,
buy: btc,
});
return;
}
let rate = match self.latest_rate.latest_rate() {
Ok(rate) => rate,
Err(e) => {
self.decline(peer, channel, Error::LatestRateFetchFailed(Box::new(e)));
return;
}
};
let xmr = match rate.sell_quote(btc) {
Ok(xmr) => xmr,
Err(e) => {
self.decline(peer, channel, Error::SellQuoteCalculationFailed(e));
return;
}
};
let xmr_balance = self.balance;
let xmr_lock_fees = self.lock_fee;
if xmr_balance < xmr + xmr_lock_fees {
self.decline(peer, channel, Error::BalanceTooLow { buy: btc });
return;
}
if self
.behaviour
.send_response(channel, spot_price::Response::Xmr(xmr))
.is_err()
{
tracing::error!(%peer, "Failed to send spot price response of {} for {}", xmr, btc)
}
self.events
.push_back(OutEvent::ExecutionSetupParams { peer, btc, xmr });
}
}
impl From<OutEvent> for alice::OutEvent {
fn from(event: OutEvent) -> Self {
match event {
OutEvent::ExecutionSetupParams { peer, btc, xmr } => {
Self::ExecutionSetupStart { peer, btc, xmr }
}
OutEvent::Error { peer, error } => Self::SwapRequestDeclined { peer, error },
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("ASB is running in resume-only mode")]
ResumeOnlyMode,
#[error("Maximum buy {max} exceeded {buy}")]
MaxBuyAmountExceeded {
max: bitcoin::Amount,
buy: bitcoin::Amount,
},
#[error("This seller's XMR balance is currently too low to fulfill the swap request to buy {buy}, please try again later")]
BalanceTooLow { buy: bitcoin::Amount },
#[error("Failed to fetch latest rate")]
LatestRateFetchFailed(#[source] Box<dyn std::error::Error + Send + 'static>),
#[error("Failed to calculate quote: {0}")]
SellQuoteCalculationFailed(#[source] anyhow::Error),
}
impl Error {
pub fn to_error_response(&self) -> spot_price::Error {
match self {
Error::ResumeOnlyMode => spot_price::Error::NoSwapsAccepted,
Error::MaxBuyAmountExceeded { max, buy } => spot_price::Error::MaxBuyAmountExceeded {
max: *max,
buy: *buy,
},
Error::BalanceTooLow { buy } => spot_price::Error::BalanceTooLow { buy: *buy },
Error::LatestRateFetchFailed(_) | Error::SellQuoteCalculationFailed(_) => {
spot_price::Error::Other
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::asb::Rate;
use crate::monero;
use crate::network::test::{await_events_or_timeout, connect, new_swarm};
use crate::protocol::{alice, bob};
use anyhow::anyhow;
use libp2p::Swarm;
use rust_decimal::Decimal;
impl Default for AliceBehaviourValues {
fn default() -> Self {
Self {
balance: monero::Amount::from_monero(1.0).unwrap(),
lock_fee: monero::Amount::ZERO,
max_buy: bitcoin::Amount::from_btc(0.01).unwrap(),
rate: TestRate::default(), // 0.01
resume_only: false,
}
}
}
#[tokio::test]
async fn given_alice_has_sufficient_balance_then_returns_price() {
let mut test = SpotPriceTest::setup(AliceBehaviourValues::default()).await;
let btc_to_swap = bitcoin::Amount::from_btc(0.01).unwrap();
let expected_xmr = monero::Amount::from_monero(1.0).unwrap();
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_price((btc_to_swap, expected_xmr), expected_xmr)
.await;
}
#[tokio::test]
async fn given_alice_has_insufficient_balance_then_returns_error() {
let mut test = SpotPriceTest::setup(
AliceBehaviourValues::default().with_balance(monero::Amount::ZERO),
)
.await;
let btc_to_swap = bitcoin::Amount::from_btc(0.01).unwrap();
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_error(
alice::spot_price::Error::BalanceTooLow { buy: btc_to_swap },
bob::spot_price::Error::BalanceTooLow { buy: btc_to_swap },
)
.await;
}
#[tokio::test]
async fn given_alice_has_insufficient_balance_after_balance_update_then_returns_error() {
let mut test = SpotPriceTest::setup(AliceBehaviourValues::default()).await;
let btc_to_swap = bitcoin::Amount::from_btc(0.01).unwrap();
let expected_xmr = monero::Amount::from_monero(1.0).unwrap();
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_price((btc_to_swap, expected_xmr), expected_xmr)
.await;
test.alice_swarm
.behaviour_mut()
.update_balance(monero::Amount::ZERO);
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_error(
alice::spot_price::Error::BalanceTooLow { buy: btc_to_swap },
bob::spot_price::Error::BalanceTooLow { buy: btc_to_swap },
)
.await;
}
#[tokio::test]
async fn given_alice_has_insufficient_balance_because_of_lock_fee_then_returns_error() {
let mut test = SpotPriceTest::setup(
AliceBehaviourValues::default().with_lock_fee(monero::Amount::from_piconero(1)),
)
.await;
let btc_to_swap = bitcoin::Amount::from_btc(0.01).unwrap();
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_error(
alice::spot_price::Error::BalanceTooLow { buy: btc_to_swap },
bob::spot_price::Error::BalanceTooLow { buy: btc_to_swap },
)
.await;
}
#[tokio::test]
async fn given_max_buy_exceeded_then_returns_error() {
let max_buy = bitcoin::Amount::from_btc(0.001).unwrap();
let mut test =
SpotPriceTest::setup(AliceBehaviourValues::default().with_max_buy(max_buy)).await;
let btc_to_swap = bitcoin::Amount::from_btc(0.01).unwrap();
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_error(
alice::spot_price::Error::MaxBuyAmountExceeded {
buy: btc_to_swap,
max: max_buy,
},
bob::spot_price::Error::MaxBuyAmountExceeded {
buy: btc_to_swap,
max: max_buy,
},
)
.await;
}
#[tokio::test]
async fn given_alice_in_resume_only_mode_then_returns_error() {
let mut test =
SpotPriceTest::setup(AliceBehaviourValues::default().with_resume_only(true)).await;
let btc_to_swap = bitcoin::Amount::from_btc(0.01).unwrap();
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_error(
alice::spot_price::Error::ResumeOnlyMode,
bob::spot_price::Error::NoSwapsAccepted,
)
.await;
}
#[tokio::test]
async fn given_rate_fetch_problem_then_returns_error() {
let mut test =
SpotPriceTest::setup(AliceBehaviourValues::default().with_rate(TestRate::error_rate()))
.await;
let btc_to_swap = bitcoin::Amount::from_btc(0.01).unwrap();
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_error(
alice::spot_price::Error::LatestRateFetchFailed(Box::new(TestRateError {})),
bob::spot_price::Error::Other,
)
.await;
}
#[tokio::test]
async fn given_rate_calculation_problem_then_returns_error() {
let mut test = SpotPriceTest::setup(
AliceBehaviourValues::default().with_rate(TestRate::from_rate_and_spread(0.0, 0)),
)
.await;
let btc_to_swap = bitcoin::Amount::from_btc(0.01).unwrap();
let request = spot_price::Request { btc: btc_to_swap };
test.send_request(request);
test.assert_error(
alice::spot_price::Error::SellQuoteCalculationFailed(anyhow!(
"Error text irrelevant, won't be checked here"
)),
bob::spot_price::Error::Other,
)
.await;
}
struct SpotPriceTest {
alice_swarm: Swarm<alice::spot_price::Behaviour<TestRate>>,
bob_swarm: Swarm<spot_price::Behaviour>,
alice_peer_id: PeerId,
}
impl SpotPriceTest {
pub async fn setup(values: AliceBehaviourValues) -> Self {
let (mut alice_swarm, _, alice_peer_id) = new_swarm(|_, _| {
Behaviour::new(
values.balance,
values.lock_fee,
values.max_buy,
values.rate.clone(),
values.resume_only,
)
});
let (mut bob_swarm, ..) = new_swarm(|_, _| bob::spot_price::bob());
connect(&mut alice_swarm, &mut bob_swarm).await;
Self {
alice_swarm,
bob_swarm,
alice_peer_id,
}
}
pub fn send_request(&mut self, spot_price_request: spot_price::Request) {
self.bob_swarm
.behaviour_mut()
.send_request(&self.alice_peer_id, spot_price_request);
}
async fn assert_price(
&mut self,
alice_assert: (bitcoin::Amount, monero::Amount),
bob_assert: monero::Amount,
) {
match await_events_or_timeout(self.alice_swarm.next(), self.bob_swarm.next()).await {
(
alice::spot_price::OutEvent::ExecutionSetupParams { btc, xmr, .. },
spot_price::OutEvent::Message { message, .. },
) => {
assert_eq!(alice_assert, (btc, xmr));
let response = match message {
RequestResponseMessage::Response { response, .. } => response,
_ => panic!("Unexpected message {:?} for Bob", message),
};
match response {
spot_price::Response::Xmr(xmr) => {
assert_eq!(bob_assert, xmr)
}
_ => panic!("Unexpected response {:?} for Bob", response),
}
}
(alice_event, bob_event) => panic!(
"Received unexpected event, alice emitted {:?} and bob emitted {:?}",
alice_event, bob_event
),
}
}
async fn assert_error(
&mut self,
alice_assert: alice::spot_price::Error,
bob_assert: bob::spot_price::Error,
) {
match await_events_or_timeout(self.alice_swarm.next(), self.bob_swarm.next()).await {
(
alice::spot_price::OutEvent::Error { error, .. },
spot_price::OutEvent::Message { message, .. },
) => {
// TODO: Somehow make PartialEq work on Alice's spot_price::Error
match (alice_assert, error) {
(
alice::spot_price::Error::BalanceTooLow { .. },
alice::spot_price::Error::BalanceTooLow { .. },
)
| (
alice::spot_price::Error::MaxBuyAmountExceeded { .. },
alice::spot_price::Error::MaxBuyAmountExceeded { .. },
)
| (
alice::spot_price::Error::LatestRateFetchFailed(_),
alice::spot_price::Error::LatestRateFetchFailed(_),
)
| (
alice::spot_price::Error::SellQuoteCalculationFailed(_),
alice::spot_price::Error::SellQuoteCalculationFailed(_),
)
| (
alice::spot_price::Error::ResumeOnlyMode,
alice::spot_price::Error::ResumeOnlyMode,
) => {}
(alice_assert, error) => {
panic!("Expected: {:?} Actual: {:?}", alice_assert, error)
}
}
let response = match message {
RequestResponseMessage::Response { response, .. } => response,
_ => panic!("Unexpected message {:?} for Bob", message),
};
match response {
spot_price::Response::Error(error) => {
assert_eq!(bob_assert, error.into())
}
_ => panic!("Unexpected response {:?} for Bob", response),
}
}
(alice_event, bob_event) => panic!(
"Received unexpected event, alice emitted {:?} and bob emitted {:?}",
alice_event, bob_event
),
}
}
}
struct AliceBehaviourValues {
pub balance: monero::Amount,
pub lock_fee: monero::Amount,
pub max_buy: bitcoin::Amount,
pub rate: TestRate, // 0.01
pub resume_only: bool,
}
impl AliceBehaviourValues {
pub fn with_balance(mut self, balance: monero::Amount) -> AliceBehaviourValues {
self.balance = balance;
self
}
pub fn with_lock_fee(mut self, lock_fee: monero::Amount) -> AliceBehaviourValues {
self.lock_fee = lock_fee;
self
}
pub fn with_max_buy(mut self, max_buy: bitcoin::Amount) -> AliceBehaviourValues {
self.max_buy = max_buy;
self
}
pub fn with_resume_only(mut self, resume_only: bool) -> AliceBehaviourValues {
self.resume_only = resume_only;
self
}
pub fn with_rate(mut self, rate: TestRate) -> AliceBehaviourValues {
self.rate = rate;
self
}
}
#[derive(Clone, Debug)]
pub enum TestRate {
Rate(Rate),
Err(TestRateError),
}
impl TestRate {
pub const RATE: f64 = 0.01;
pub fn from_rate_and_spread(rate: f64, spread: u64) -> Self {
let ask = bitcoin::Amount::from_btc(rate).expect("Static value should never fail");
let spread = Decimal::from(spread);
Self::Rate(Rate::new(ask, spread))
}
pub fn error_rate() -> Self {
Self::Err(TestRateError {})
}
}
impl Default for TestRate {
fn default() -> Self {
TestRate::from_rate_and_spread(Self::RATE, 0)
}
}
#[derive(Debug, Clone, thiserror::Error)]
#[error("Could not fetch rate")]
pub struct TestRateError {}
impl LatestRate for TestRate {
type Error = TestRateError;
fn latest_rate(&mut self) -> Result<Rate, Self::Error> {
match self {
TestRate::Rate(rate) => Ok(*rate),
TestRate::Err(error) => Err(error.clone()),
}
}
}
}

View File

@ -16,6 +16,7 @@ pub mod cancel;
pub mod event_loop;
mod execution_setup;
pub mod refund;
pub mod spot_price;
pub mod state;
pub mod swap;

View File

@ -1,5 +1,6 @@
use crate::network::quote::BidQuote;
use crate::network::{encrypted_signature, quote, redial, spot_price, transfer_proof};
use crate::protocol::bob;
use crate::protocol::bob::{execution_setup, State2};
use anyhow::{anyhow, Error, Result};
use libp2p::core::Multiaddr;
@ -71,7 +72,7 @@ impl Behaviour {
pub fn new(alice: PeerId) -> Self {
Self {
quote: quote::bob(),
spot_price: spot_price::bob(),
spot_price: bob::spot_price::bob(),
execution_setup: Default::default(),
transfer_proof: transfer_proof::bob(),
encrypted_signature: encrypted_signature::bob(),

View File

@ -1,9 +1,11 @@
use crate::bitcoin::EncryptedSignature;
use crate::network::quote::BidQuote;
use crate::network::spot_price::Response;
use crate::network::{encrypted_signature, spot_price};
use crate::protocol::bob;
use crate::protocol::bob::{Behaviour, OutEvent, State0, State2};
use crate::{bitcoin, monero};
use anyhow::{Context, Result};
use anyhow::{bail, Context, Result};
use futures::future::{BoxFuture, OptionFuture};
use futures::{FutureExt, StreamExt};
use libp2p::request_response::{RequestId, ResponseChannel};
@ -261,11 +263,18 @@ impl EventLoopHandle {
}
pub async fn request_spot_price(&mut self, btc: bitcoin::Amount) -> Result<monero::Amount> {
Ok(self
let response = self
.spot_price
.send_receive(spot_price::Request { btc })
.await?
.xmr)
.await?;
match response {
Response::Xmr(xmr) => Ok(xmr),
Response::Error(error) => {
let error: bob::spot_price::Error = error.into();
bail!(error);
}
}
}
pub async fn request_quote(&mut self) -> Result<BidQuote> {

View File

@ -0,0 +1,69 @@
use crate::network::cbor_request_response::CborCodec;
use crate::network::spot_price;
use crate::network::spot_price::SpotPriceProtocol;
use crate::protocol::bob::OutEvent;
use libp2p::request_response::{ProtocolSupport, RequestResponseConfig};
use libp2p::PeerId;
const PROTOCOL: &str = spot_price::PROTOCOL;
pub type SpotPriceOutEvent = spot_price::OutEvent;
/// Constructs a new instance of the `spot-price` behaviour to be used by Bob.
///
/// Bob only supports outbound connections, i.e. requesting a spot price for a
/// given amount of BTC in XMR.
pub fn bob() -> spot_price::Behaviour {
spot_price::Behaviour::new(
CborCodec::default(),
vec![(SpotPriceProtocol, ProtocolSupport::Outbound)],
RequestResponseConfig::default(),
)
}
impl From<(PeerId, spot_price::Message)> for OutEvent {
fn from((peer, message): (PeerId, spot_price::Message)) -> Self {
match message {
spot_price::Message::Request { .. } => Self::unexpected_request(peer),
spot_price::Message::Response {
response,
request_id,
} => Self::SpotPriceReceived {
id: request_id,
response,
},
}
}
}
crate::impl_from_rr_event!(SpotPriceOutEvent, OutEvent, PROTOCOL);
#[derive(Clone, Debug, thiserror::Error, PartialEq)]
pub enum Error {
#[error("Seller currently does not accept incoming swap requests, please try again later")]
NoSwapsAccepted,
#[error("Seller refused to buy {buy} because the maximum configured buy limit is {max}")]
MaxBuyAmountExceeded {
max: bitcoin::Amount,
buy: bitcoin::Amount,
},
#[error("Seller's XMR balance is currently too low to fulfill the swap request to buy {buy}, please try again later")]
BalanceTooLow { buy: bitcoin::Amount },
/// To be used for errors that cannot be explained on the CLI side (e.g.
/// rate update problems on the seller side)
#[error("Seller encountered a problem, please try again later.")]
Other,
}
impl From<spot_price::Error> for Error {
fn from(error: spot_price::Error) -> Self {
match error {
spot_price::Error::NoSwapsAccepted => Error::NoSwapsAccepted,
spot_price::Error::MaxBuyAmountExceeded { max, buy } => {
Error::MaxBuyAmountExceeded { max, buy }
}
spot_price::Error::BalanceTooLow { buy } => Error::BalanceTooLow { buy },
spot_price::Error::Other => Error::Other,
}
}
}

View File

@ -90,7 +90,8 @@ where
env_config,
alice_bitcoin_wallet.clone(),
alice_monero_wallet.clone(),
);
)
.await;
let bob_seed = Seed::random().unwrap();
let bob_starting_balances = StartingBalances::new(btc_amount * 10, monero::Amount::ZERO, None);
@ -213,7 +214,7 @@ pub async fn init_electrs_container(
Ok(docker)
}
fn start_alice(
async fn start_alice(
seed: &Seed,
db_path: PathBuf,
listen_address: Multiaddr,
@ -223,7 +224,21 @@ fn start_alice(
) -> (AliceApplicationHandle, Receiver<alice::Swap>) {
let db = Arc::new(Database::open(db_path.as_path()).unwrap());
let mut swarm = swarm::alice(&seed).unwrap();
let current_balance = monero_wallet.get_balance().await.unwrap();
let lock_fee = monero_wallet.static_tx_fee_estimate();
let max_buy = bitcoin::Amount::from_sat(u64::MAX);
let latest_rate = FixedRate::default();
let resume_only = false;
let mut swarm = swarm::alice(
&seed,
current_balance,
lock_fee,
max_buy,
latest_rate,
resume_only,
)
.unwrap();
swarm.listen_on(listen_address).unwrap();
let (event_loop, swap_handle) = alice::EventLoop::new(
@ -496,7 +511,8 @@ impl TestContext {
self.env_config,
self.alice_bitcoin_wallet.clone(),
self.alice_monero_wallet.clone(),
);
)
.await;
self.alice_handle = alice_handle;
self.alice_swap_handle = alice_swap_handle;