203: Introduce dynamic rates r=da-kami a=D4nte



Co-authored-by: Franck Royer <franck@coblox.tech>
Co-authored-by: Thomas Eizinger <thomas@eizinger.io>
Co-authored-by: Daniel Karzel <daniel@comit.network>
This commit is contained in:
bors[bot] 2021-02-22 04:36:49 +00:00 committed by GitHub
commit 61a8282be1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 411 additions and 33 deletions

64
Cargo.lock generated
View File

@ -1504,6 +1504,15 @@ dependencies = [
"hashbrown",
]
[[package]]
name = "input_buffer"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f97967975f448f1a7ddb12b0bc41069d09ed6a1c161a92687e057325db35d413"
dependencies = [
"bytes",
]
[[package]]
name = "instant"
version = "0.1.9"
@ -3163,6 +3172,19 @@ dependencies = [
"serde",
]
[[package]]
name = "sha-1"
version = "0.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dfebf75d25bd900fd1e7d11501efab59bc846dbc76196839663e6637bba9f25f"
dependencies = [
"block-buffer 0.9.0",
"cfg-if 1.0.0",
"cpuid-bool 0.1.2",
"digest 0.9.0",
"opaque-debug 0.3.0",
]
[[package]]
name = "sha1"
version = "0.6.0"
@ -3501,6 +3523,7 @@ dependencies = [
"thiserror",
"time",
"tokio",
"tokio-tungstenite",
"toml",
"tracing",
"tracing-core",
@ -3749,6 +3772,21 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-tungstenite"
version = "0.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1a5f475f1b9d077ea1017ecbc60890fda8e54942d680ca0b1d2b47cfa2d861b"
dependencies = [
"futures-util",
"log",
"native-tls",
"pin-project 1.0.4",
"tokio",
"tokio-native-tls",
"tungstenite",
]
[[package]]
name = "tokio-util"
version = "0.6.1"
@ -3856,6 +3894,26 @@ version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "59547bce71d9c38b83d9c0e92b6066c4253371f15005def0c30d9657f50c7642"
[[package]]
name = "tungstenite"
version = "0.12.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8ada8297e8d70872fa9a551d93250a9f407beb9f37ef86494eb20012a2ff7c24"
dependencies = [
"base64 0.13.0",
"byteorder",
"bytes",
"http",
"httparse",
"input_buffer",
"log",
"native-tls",
"rand 0.8.2",
"sha-1",
"url",
"utf-8",
]
[[package]]
name = "typenum"
version = "1.12.0"
@ -3969,6 +4027,12 @@ dependencies = [
"serde",
]
[[package]]
name = "utf-8"
version = "0.7.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05e42f7c18b8f902290b009cde6d651262f956c98bc51bca4cd1d511c9cd85c7"
[[package]]
name = "uuid"
version = "0.8.2"

View File

@ -55,6 +55,7 @@ tempfile = "3"
thiserror = "1"
time = "0.2"
tokio = { version = "1.0", features = ["rt-multi-thread", "time", "macros", "sync"] }
tokio-tungstenite = { version = "0.13", features = [ "tls" ] }
toml = "0.5"
tracing = { version = "0.1", features = ["attributes"] }
tracing-core = "0.1"

View File

@ -1,2 +1,14 @@
pub mod command;
pub mod config;
pub mod fixed_rate;
pub mod kraken;
mod amounts;
pub use amounts::Rate;
pub trait LatestRate {
type Error: std::error::Error + Send + Sync + 'static;
fn latest_rate(&mut self) -> Result<Rate, Self::Error>;
}

73
swap/src/asb/amounts.rs Normal file
View File

@ -0,0 +1,73 @@
use crate::{bitcoin, monero};
use anyhow::{anyhow, Result};
use rust_decimal::{prelude::ToPrimitive, Decimal};
use std::fmt::{Debug, Display, Formatter};
/// Prices at which 1 XMR will be traded, in BTC (XMR/BTC pair)
/// The `ask` represents the minimum price in BTC for which we are willing to
/// sell 1 XMR.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Rate {
pub ask: bitcoin::Amount,
}
impl Rate {
pub const ZERO: Rate = Rate {
ask: bitcoin::Amount::ZERO,
};
// This function takes the quote amount as it is what Bob sends to Alice in the
// swap request
pub fn sell_quote(&self, quote: bitcoin::Amount) -> Result<monero::Amount> {
Self::quote(self.ask, quote)
}
fn quote(rate: bitcoin::Amount, quote: bitcoin::Amount) -> Result<monero::Amount> {
// quote (btc) = rate * base (xmr)
// base = quote / rate
let quote_in_sats = quote.as_sat();
let quote_in_btc = Decimal::from(quote_in_sats)
.checked_div(Decimal::from(bitcoin::Amount::ONE_BTC.as_sat()))
.ok_or_else(|| anyhow!("division overflow"))?;
let rate_in_btc = Decimal::from(rate.as_sat())
.checked_div(Decimal::from(bitcoin::Amount::ONE_BTC.as_sat()))
.ok_or_else(|| anyhow!("division overflow"))?;
let base_in_xmr = quote_in_btc
.checked_div(rate_in_btc)
.ok_or_else(|| anyhow!("division overflow"))?;
let base_in_piconero = base_in_xmr * Decimal::from(monero::Amount::ONE_XMR.as_piconero());
let base_in_piconero = base_in_piconero
.to_u64()
.ok_or_else(|| anyhow!("decimal cannot be represented as u64"))?;
Ok(monero::Amount::from_piconero(base_in_piconero))
}
}
impl Display for Rate {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.ask)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sell_quote() {
let rate = Rate {
ask: bitcoin::Amount::from_btc(0.002_500).unwrap(),
};
let btc_amount = bitcoin::Amount::from_btc(2.5).unwrap();
let xmr_amount = rate.sell_quote(btc_amount).unwrap();
assert_eq!(xmr_amount, monero::Amount::from_monero(1000.0).unwrap())
}
}

View File

@ -0,0 +1,23 @@
use crate::asb::{LatestRate, Rate};
use std::convert::Infallible;
pub const RATE: f64 = 0.01;
#[derive(Clone)]
pub struct RateService(Rate);
impl LatestRate for RateService {
type Error = Infallible;
fn latest_rate(&mut self) -> Result<Rate, Infallible> {
Ok(self.0)
}
}
impl Default for RateService {
fn default() -> Self {
Self(Rate {
ask: bitcoin::Amount::from_btc(RATE).expect("Static value should never fail"),
})
}
}

181
swap/src/asb/kraken.rs Normal file
View File

@ -0,0 +1,181 @@
use crate::asb::{LatestRate, Rate};
use anyhow::Result;
use bitcoin::util::amount::ParseAmountError;
use futures::{SinkExt, StreamExt};
use reqwest::Url;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::convert::TryFrom;
use tokio::sync::watch;
use tokio_tungstenite::tungstenite::Message;
use watch::Receiver;
const KRAKEN_WS_URL: &str = "wss://ws.kraken.com";
const SUBSCRIBE_XMR_BTC_TICKER_PAYLOAD: &str = r#"
{ "event": "subscribe",
"pair": [ "XMR/XBT" ],
"subscription": {
"name": "ticker"
}
}"#;
#[derive(Clone)]
pub struct RateService {
receiver: Receiver<Result<Rate, Error>>,
}
impl LatestRate for RateService {
type Error = Error;
fn latest_rate(&mut self) -> Result<Rate, Self::Error> {
(*self.receiver.borrow()).clone()
}
}
#[derive(Clone, Debug, thiserror::Error)]
pub enum Error {
#[error("Rate has not yet been retrieved from Kraken websocket API")]
NotYetRetrieved,
#[error("Message is not text")]
NonTextMessage,
#[error("Websocket: ")]
WebSocket(String),
#[error("Serde: ")]
Serde(String),
#[error("Data field is missing")]
DataFieldMissing,
#[error("Ask Rate Element is of unexpected type")]
UnexpectedAskRateElementType,
#[error("Ask Rate Element is missing")]
MissingAskRateElementType,
#[error("Bitcoin amount parse error: ")]
BitcoinParseAmount(#[from] ParseAmountError),
}
impl From<tokio_tungstenite::tungstenite::Error> for Error {
fn from(err: tokio_tungstenite::tungstenite::Error) -> Self {
Error::WebSocket(format!("{:#}", err))
}
}
impl From<serde_json::Error> for Error {
fn from(err: serde_json::Error) -> Self {
Error::Serde(format!("{:#}", err))
}
}
impl RateService {
pub async fn new() -> Result<Self> {
let (tx, rx) = watch::channel(Err(Error::NotYetRetrieved));
let (ws, _response) =
tokio_tungstenite::connect_async(Url::parse(KRAKEN_WS_URL).expect("valid url")).await?;
let (mut write, mut read) = ws.split();
tokio::spawn(async move {
while let Some(msg) = read.next().await {
let msg = match msg {
Ok(Message::Text(msg)) => msg,
Ok(_) => {
let _ = tx.send(Err(Error::NonTextMessage));
continue;
}
Err(e) => {
let _ = tx.send(Err(e.into()));
continue;
}
};
// If we encounter a heartbeat we skip it and iterate again
if msg.eq(r#"{"event":"heartbeat"}"#) {
continue;
}
let ticker = match serde_json::from_str::<TickerUpdate>(&msg) {
Ok(ticker) => ticker,
Err(e) => {
let _ = tx.send(Err(e.into()));
continue;
}
};
let rate = match Rate::try_from(ticker) {
Ok(rate) => rate,
Err(e) => {
let _ = tx.send(Err(e));
continue;
}
};
let _ = tx.send(Ok(rate));
}
});
write.send(SUBSCRIBE_XMR_BTC_TICKER_PAYLOAD.into()).await?;
Ok(Self { receiver: rx })
}
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(transparent)]
struct TickerUpdate(Vec<TickerField>);
#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)]
enum TickerField {
Data(TickerData),
Metadata(Value),
}
#[derive(Debug, Serialize, Deserialize)]
struct TickerData {
#[serde(rename = "a")]
ask: Vec<RateElement>,
#[serde(rename = "b")]
bid: Vec<RateElement>,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)]
enum RateElement {
Text(String),
Number(u64),
}
impl TryFrom<TickerUpdate> for Rate {
type Error = Error;
fn try_from(value: TickerUpdate) -> Result<Self, Error> {
let data = value
.0
.iter()
.find_map(|field| match field {
TickerField::Data(data) => Some(data),
TickerField::Metadata(_) => None,
})
.ok_or(Error::DataFieldMissing)?;
let ask = data.ask.first().ok_or(Error::MissingAskRateElementType)?;
let ask = match ask {
RateElement::Text(ask) => {
bitcoin::Amount::from_str_in(ask, ::bitcoin::Denomination::Bitcoin)?
}
_ => return Err(Error::UnexpectedAskRateElementType),
};
Ok(Self { ask })
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn deserialize_ticker_update() {
let sample_response = r#"[980,{"a":["0.00521900",4,"4.84775132"],"b":["0.00520600",70,"70.35668921"],"c":["0.00520700","0.00000186"],"v":["18530.40510860","18531.94887860"],"p":["0.00489493","0.00489490"],"t":[5017,5018],"l":["0.00448300","0.00448300"],"h":["0.00525000","0.00525000"],"o":["0.00450000","0.00451000"]},"ticker","XMR/XBT"]"#;
let _ = serde_json::from_str::<TickerUpdate>(sample_response).unwrap();
}
}

View File

@ -24,6 +24,7 @@ use swap::{
initial_setup, query_user_for_initial_testnet_config, read_config, Config,
ConfigNotInitialized,
},
kraken,
},
bitcoin,
database::Database,
@ -96,6 +97,8 @@ async fn main() -> Result<()> {
bitcoin_wallet.new_address().await?
);
let rate_service = kraken::RateService::new().await?;
let (mut event_loop, _) = EventLoop::new(
config.network.listen,
seed,
@ -103,6 +106,7 @@ async fn main() -> Result<()> {
Arc::new(bitcoin_wallet),
Arc::new(monero_wallet),
Arc::new(db),
rate_service,
)
.unwrap();

View File

@ -283,7 +283,7 @@ pub async fn current_epoch<W>(
cancel_timelock: CancelTimelock,
punish_timelock: PunishTimelock,
lock_tx_id: ::bitcoin::Txid,
) -> anyhow::Result<ExpiredTimelocks>
) -> Result<ExpiredTimelocks>
where
W: WatchForRawTransaction + TransactionBlockHeight + GetBlockHeight,
{

View File

@ -1,4 +1,5 @@
use crate::bitcoin;
use anyhow::Result;
use libp2p::{core::Multiaddr, PeerId};
use std::path::PathBuf;
use uuid::Uuid;
@ -85,7 +86,7 @@ pub enum Refund {
},
}
fn parse_btc(str: &str) -> anyhow::Result<bitcoin::Amount> {
fn parse_btc(str: &str) -> Result<bitcoin::Amount> {
let amount = bitcoin::Amount::from_str_in(str, ::bitcoin::Denomination::Bitcoin)?;
Ok(amount)
}

View File

@ -65,7 +65,7 @@ impl Database {
.context("Could not flush db")
}
pub fn get_state(&self, swap_id: Uuid) -> anyhow::Result<Swap> {
pub fn get_state(&self, swap_id: Uuid) -> Result<Swap> {
let key = serialize(&swap_id)?;
let encoded = self
@ -97,14 +97,14 @@ impl Database {
}
}
pub fn serialize<T>(t: &T) -> anyhow::Result<Vec<u8>>
pub fn serialize<T>(t: &T) -> Result<Vec<u8>>
where
T: Serialize,
{
Ok(serde_cbor::to_vec(t)?)
}
pub fn deserialize<T>(v: &[u8]) -> anyhow::Result<T>
pub fn deserialize<T>(v: &[u8]) -> Result<T>
where
T: DeserializeOwned,
{

View File

@ -1,4 +1,4 @@
use anyhow::Context;
use anyhow::{Context, Result};
use directories_next::ProjectDirs;
use std::path::{Path, PathBuf};
@ -9,7 +9,7 @@ fn default_config_dir() -> Option<PathBuf> {
ProjectDirs::from("", "", "xmr-btc-swap").map(|proj_dirs| proj_dirs.config_dir().to_path_buf())
}
pub fn default_config_path() -> anyhow::Result<PathBuf> {
pub fn default_config_path() -> Result<PathBuf> {
default_config_dir()
.map(|dir| Path::join(&dir, "config.toml"))
.context("Could not generate default configuration path")

View File

@ -77,6 +77,7 @@ pub struct Amount(u64);
impl Amount {
pub const ZERO: Self = Self(0);
pub const ONE_XMR: Self = Self(PICONERO_OFFSET);
/// Create an [Amount] with piconero precision and the given number of
/// piconeros.
///
@ -185,7 +186,7 @@ pub trait Transfer {
public_spend_key: PublicKey,
public_view_key: PublicViewKey,
amount: Amount,
) -> anyhow::Result<(TransferProof, Amount)>;
) -> Result<(TransferProof, Amount)>;
}
#[async_trait]
@ -214,17 +215,17 @@ pub trait CreateWalletForOutput {
private_spend_key: PrivateKey,
private_view_key: PrivateViewKey,
restore_height: Option<u32>,
) -> anyhow::Result<()>;
) -> Result<()>;
}
#[async_trait]
pub trait OpenWallet {
async fn open_wallet(&self, file_name: &str) -> anyhow::Result<()>;
async fn open_wallet(&self, file_name: &str) -> Result<()>;
}
#[async_trait]
pub trait CreateWallet {
async fn create_wallet(&self, file_name: &str) -> anyhow::Result<()>;
async fn create_wallet(&self, file_name: &str) -> Result<()>;
}
#[derive(thiserror::Error, Debug, Clone, PartialEq)]

View File

@ -119,7 +119,7 @@ impl Behaviour {
&mut self,
channel: ResponseChannel<QuoteResponse>,
quote_response: QuoteResponse,
) -> anyhow::Result<()> {
) -> Result<()> {
self.quote_response.send(channel, quote_response)?;
info!("Sent quote response");
Ok(())
@ -127,7 +127,6 @@ impl Behaviour {
pub fn start_execution_setup(&mut self, bob_peer_id: PeerId, state0: State0) {
self.execution_setup.run(bob_peer_id, state0);
info!("Start execution setup with {}", bob_peer_id);
}
/// Send Transfer Proof to Bob.

View File

@ -1,4 +1,5 @@
use crate::{
asb::LatestRate,
bitcoin,
database::Database,
execution_params::ExecutionParams,
@ -21,12 +22,9 @@ use libp2p::{
use rand::rngs::OsRng;
use std::sync::Arc;
use tokio::sync::{broadcast, mpsc, mpsc::error::SendError};
use tracing::{debug, error, trace};
use tracing::{debug, error, info, trace};
use uuid::Uuid;
// TODO: Use dynamic
pub const RATE: u32 = 100;
#[allow(missing_debug_implementations)]
pub struct MpscChannels<T> {
sender: mpsc::Sender<T>,
@ -79,7 +77,7 @@ impl EventLoopHandle {
}
#[allow(missing_debug_implementations)]
pub struct EventLoop {
pub struct EventLoop<RS> {
swarm: libp2p::Swarm<Behaviour>,
peer_id: PeerId,
execution_params: ExecutionParams,
@ -87,6 +85,7 @@ pub struct EventLoop {
monero_wallet: Arc<monero::Wallet>,
db: Arc<Database>,
listen_address: Multiaddr,
rate_service: RS,
recv_encrypted_signature: broadcast::Sender<EncryptedSignature>,
send_transfer_proof: mpsc::Receiver<(PeerId, TransferProof)>,
@ -97,7 +96,10 @@ pub struct EventLoop {
swap_handle_sender: mpsc::Sender<RemoteHandle<Result<AliceState>>>,
}
impl EventLoop {
impl<RS> EventLoop<RS>
where
RS: LatestRate,
{
pub fn new(
listen_address: Multiaddr,
seed: Seed,
@ -105,6 +107,7 @@ impl EventLoop {
bitcoin_wallet: Arc<bitcoin::Wallet>,
monero_wallet: Arc<monero::Wallet>,
db: Arc<Database>,
rate_service: RS,
) -> Result<(Self, mpsc::Receiver<RemoteHandle<Result<AliceState>>>)> {
let identity = network::Seed::new(seed).derive_libp2p_identity();
let behaviour = Behaviour::default();
@ -132,6 +135,7 @@ impl EventLoop {
monero_wallet,
db,
listen_address,
rate_service,
recv_encrypted_signature: recv_encrypted_signature.sender,
send_transfer_proof: send_transfer_proof.receiver,
send_transfer_proof_sender: send_transfer_proof.sender,
@ -160,7 +164,9 @@ impl EventLoop {
debug!("Connection Established with {}", alice);
}
OutEvent::QuoteRequest { msg, channel, bob_peer_id } => {
let _ = self.handle_quote_request(msg, channel, bob_peer_id).await;
if let Err(error) = self.handle_quote_request(msg, channel, bob_peer_id).await {
error!("Failed to handle quote request: {:#}", error);
}
}
OutEvent::ExecutionSetupDone{bob_peer_id, state3} => {
let _ = self.handle_execution_setup_done(bob_peer_id, *state3).await;
@ -199,9 +205,13 @@ impl EventLoop {
// 1. Check if acceptable request
// 2. Send response
let rate = self
.rate_service
.latest_rate()
.context("Failed to get latest rate")?;
let btc_amount = quote_request.btc_amount;
let xmr_amount = btc_amount.as_btc() * RATE as f64;
let xmr_amount = monero::Amount::from_monero(xmr_amount)?;
let xmr_amount = rate.sell_quote(btc_amount)?;
let quote_response = QuoteResponse { xmr_amount };
self.swarm
@ -219,6 +229,11 @@ impl EventLoop {
)
.await?;
info!(
"Starting execution setup to sell {} for {} (rate of {}) with {}",
xmr_amount, btc_amount, rate, bob_peer_id
);
self.swarm.start_execution_setup(bob_peer_id, state0);
// Continues once the execution setup protocol is done
Ok(())

View File

@ -140,7 +140,7 @@ impl State0 {
}
}
pub async fn receive<W>(self, wallet: &W, msg: Message1) -> anyhow::Result<State1>
pub async fn receive<W>(self, wallet: &W, msg: Message1) -> Result<State1>
where
W: BuildTxLockPsbt + GetNetwork,
{

View File

@ -1,10 +1,11 @@
use anyhow::Result;
use atty::{self};
use log::LevelFilter;
use tracing::{info, subscriber};
use tracing_log::LogTracer;
use tracing_subscriber::FmtSubscriber;
pub fn init_tracing(level: LevelFilter) -> anyhow::Result<()> {
pub fn init_tracing(level: LevelFilter) -> Result<()> {
if level == LevelFilter::Off {
return Ok(());
}

View File

@ -14,18 +14,14 @@ use std::{
time::Duration,
};
use swap::{
asb::{fixed_rate, fixed_rate::RATE},
bitcoin,
bitcoin::{CancelTimelock, PunishTimelock},
database::Database,
execution_params,
execution_params::{ExecutionParams, GetExecutionParams},
monero,
protocol::{
alice,
alice::{event_loop::RATE, AliceState},
bob,
bob::BobState,
},
protocol::{alice, alice::AliceState, bob, bob::BobState},
seed::Seed,
};
use tempfile::tempdir;
@ -88,7 +84,7 @@ pub struct TestContext {
alice_starting_balances: StartingBalances,
alice_bitcoin_wallet: Arc<bitcoin::Wallet>,
alice_monero_wallet: Arc<monero::Wallet>,
alice_swap_handle: mpsc::Receiver<RemoteHandle<anyhow::Result<AliceState>>>,
alice_swap_handle: mpsc::Receiver<RemoteHandle<Result<AliceState>>>,
bob_params: BobParams,
bob_starting_balances: StartingBalances,
@ -148,7 +144,13 @@ impl TestContext {
.get_balance()
.await
.unwrap();
assert!(xmr_balance_after_swap <= self.alice_starting_balances.xmr - self.xmr_amount);
assert!(
xmr_balance_after_swap <= self.alice_starting_balances.xmr - self.xmr_amount,
"{} !< {} - {}",
xmr_balance_after_swap,
self.alice_starting_balances.xmr,
self.xmr_amount
);
}
pub async fn assert_alice_refunded(&mut self) {
@ -324,7 +326,7 @@ where
let (monero, containers) = testutils::init_containers(&cli).await;
let btc_amount = bitcoin::Amount::from_sat(1_000_000);
let xmr_amount = monero::Amount::from_monero(btc_amount.as_btc() * RATE as f64).unwrap();
let xmr_amount = monero::Amount::from_monero(btc_amount.as_btc() / RATE).unwrap();
let alice_starting_balances = StartingBalances {
xmr: xmr_amount * 10,
@ -390,6 +392,7 @@ where
alice_bitcoin_wallet.clone(),
alice_monero_wallet.clone(),
alice_db,
fixed_rate::RateService::default(),
)
.unwrap();