refactor: swap-feed crate (#464)

* refactor: Move price feed of asb into swap-feed crate

Also move the tracing-subscriber deps into the workspace

* fix conversion methods

Cargo.lock

@@ -9822,6 +9822,7 @@ dependencies = [
  "structopt",
  "strum 0.26.3",
  "swap-env",
+ "swap-feed",
  "swap-fs",
  "swap-serde",
  "tauri",
@@ -9870,6 +9871,26 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "swap-feed"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "backoff",
+ "bitcoin 0.32.6",
+ "futures",
+ "monero",
+ "rust_decimal",
+ "serde",
+ "serde_json",
+ "thiserror 1.0.69",
+ "tokio",
+ "tokio-tungstenite",
+ "tracing",
+ "tracing-subscriber",
+ "url",
+]
+
 [[package]]
 name = "swap-fs"
 version = "0.1.0"

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = [ "electrum-pool", "monero-rpc", "monero-rpc-pool", "monero-sys", "monero-seed", "src-tauri", "swap", "swap-env", "swap-fs", "swap-serde"]
+members = [ "electrum-pool", "monero-rpc", "monero-rpc-pool", "monero-sys", "monero-seed", "src-tauri", "swap", "swap-env", "swap-fs", "swap-feed", "swap-serde"]
 
 [workspace.dependencies]
 anyhow = "1"
@@ -9,6 +9,7 @@ serde_json = "1"
 tokio = { version = "1", features = ["rt-multi-thread", "time", "macros", "sync"] }
 futures = { version = "0.3", default-features = false, features = ["std"] }
 tracing = { version = "0.1", features = ["attributes"] }
+tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt", "ansi", "env-filter", "time", "tracing-log", "json"] }
 bitcoin = { version = "0.32", features = ["rand", "serde"] }
 monero = { version = "0.12", features = ["serde_support"] }
 rand = "0.8"

monero-harness/Cargo.toml

@@ -16,4 +16,4 @@ reqwest = { workspace = true }
 testcontainers = "0.15"
 tokio = { workspace = true, features = ["rt-multi-thread", "time", "macros"] }
 tracing = { workspace = true }
-tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt", "ansi", "env-filter", "tracing-log"] }
+tracing-subscriber = { workspace = true }

monero-rpc-pool/Cargo.toml

@@ -26,7 +26,7 @@ tokio = { workspace = true, features = ["full"] }
 tower = "0.4"
 tower-http = { version = "0.5", features = ["cors"] }
 tracing = { workspace = true }
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-subscriber = { workspace = true }
 typeshare = { workspace = true }
 url = "2.0"
 uuid = { workspace = true }

monero-sys/Cargo.toml

@@ -29,5 +29,5 @@ quickcheck_macros = "1.0"
 tempfile = "3.19.1"
 testcontainers = "0.15"
 tokio = { workspace = true, features = ["full"] }
-tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
+tracing-subscriber = { workspace = true }
 uuid = { workspace = true }

monero-sys/monero

@@ -1 +1 @@
-Subproject commit 5f714f147fd29228698070e6bd80e41ce2f86fb0
+Subproject commit dbbccecc89e1121762a4ad6b531638ece82aa0c7

swap-feed/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "swap-feed"
+version = "0.1.0"
+authors = ["The COMIT guys <hello@comit.network>"]
+edition = "2021"
+description = "Price feed functionality for XMR/BTC atomic swaps"
+
+[lib]
+name = "swap_feed"
+
+[[bin]]
+name = "kraken_ticker"
+path = "src/bin/kraken_ticker.rs"
+
+[dependencies]
+anyhow = { workspace = true }
+backoff = { version = "0.4", features = ["tokio"] }
+bitcoin = { workspace = true }
+futures = { workspace = true }
+monero = { workspace = true }
+rust_decimal = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tokio-tungstenite = { version = "0.15", features = ["rustls-tls"] }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+url = { workspace = true }

swap-feed/src/bin/kraken_ticker.rs

@@ -9,7 +9,7 @@ async fn main() -> Result<()> {
 
     let price_ticker_ws_url = Url::parse("wss://ws.kraken.com")?;
     let mut ticker =
-        swap::kraken::connect(price_ticker_ws_url).context("Failed to connect to kraken")?;
+        swap_feed::kraken::connect(price_ticker_ws_url).context("Failed to connect to kraken")?;
 
     loop {
         match ticker.wait_for_next_update().await? {

swap-feed/src/lib.rs

@@ -0,0 +1,13 @@
+pub mod kraken;
+pub mod rate;
+pub mod traits;
+
+// Re-exports for convenience
+pub use kraken::{connect, PriceUpdates, Error as KrakenError};
+pub use rate::{Rate, FixedRate, KrakenRate};
+pub use traits::LatestRate;
+
+// Core functions
+pub fn connect_kraken(url: url::Url) -> anyhow::Result<kraken::PriceUpdates> {
+    kraken::connect(url)
+}

swap-feed/src/rate.rs

@@ -1,7 +1,7 @@
-use crate::{bitcoin, monero};
 use anyhow::{Context, Result};
 use rust_decimal::prelude::ToPrimitive;
 use rust_decimal::Decimal;
+use std::convert::Infallible;
 use std::fmt::{Debug, Display, Formatter};
 
 /// Represents the rate at which we are willing to trade 1 XMR.
@@ -63,13 +63,13 @@ impl Rate {
         let base_in_xmr = quote_in_btc
             .checked_div(rate_in_btc)
             .context("Division overflow")?;
-        let base_in_piconero = base_in_xmr * Decimal::from(monero::Amount::ONE_XMR.as_piconero());
+        let base_in_piconero = base_in_xmr * Decimal::from(monero::Amount::ONE_XMR.as_pico());
 
         let base_in_piconero = base_in_piconero
             .to_u64()
             .context("Failed to fit piconero amount into a u64")?;
 
-        Ok(monero::Amount::from_piconero(base_in_piconero))
+        Ok(monero::Amount::from_pico(base_in_piconero))
     }
 }
 
@@ -79,6 +79,62 @@ impl Display for Rate {
     }
 }
 
+#[derive(Clone, Debug)]
+pub struct FixedRate(Rate);
+
+impl FixedRate {
+    pub const RATE: f64 = 0.01;
+
+    pub fn value(&self) -> Rate {
+        self.0
+    }
+}
+
+impl Default for FixedRate {
+    fn default() -> Self {
+        let ask = bitcoin::Amount::from_btc(Self::RATE).expect("Static value should never fail");
+        let spread = Decimal::from(0u64);
+
+        Self(Rate::new(ask, spread))
+    }
+}
+
+impl crate::traits::LatestRate for FixedRate {
+    type Error = Infallible;
+
+    fn latest_rate(&mut self) -> Result<Rate, Self::Error> {
+        Ok(self.value())
+    }
+}
+
+/// Produces [`Rate`]s based on [`PriceUpdate`]s from kraken and a configured
+/// spread.
+#[derive(Debug, Clone)]
+pub struct KrakenRate {
+    ask_spread: Decimal,
+    price_updates: crate::kraken::PriceUpdates,
+}
+
+impl KrakenRate {
+    pub fn new(ask_spread: Decimal, price_updates: crate::kraken::PriceUpdates) -> Self {
+        Self {
+            ask_spread,
+            price_updates,
+        }
+    }
+}
+
+impl crate::traits::LatestRate for KrakenRate {
+    type Error = crate::kraken::Error;
+
+    fn latest_rate(&mut self) -> Result<Rate, Self::Error> {
+        let update = self.price_updates.latest_update()?;
+        let rate = Rate::new(update.ask, self.ask_spread);
+
+        Ok(rate)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -95,7 +151,7 @@ mod tests {
 
         let xmr_amount = rate.sell_quote(btc_amount).unwrap();
 
-        assert_eq!(xmr_amount, monero::Amount::from_monero(1000.0).unwrap())
+        assert_eq!(xmr_amount, monero::Amount::from_xmr(1000.0).unwrap())
     }
 
     #[test]
@@ -122,7 +178,7 @@ mod tests {
             .unwrap();
 
         let xmr_factor =
-            xmr_no_spread.as_piconero_decimal() / xmr_with_spread.as_piconero_decimal() - ONE;
+            xmr_no_spread.into().as_piconero_decimal() / xmr_with_spread.into().as_piconero_decimal() - ONE;
 
         assert!(xmr_with_spread < xmr_no_spread);
         assert_eq!(xmr_factor.round_dp(8), TWO_PERCENT); // round to 8 decimal
@@ -130,4 +186,4 @@ mod tests {
                                                          // it is really close
                                                          // to two percent
     }
-}
+}

swap-feed/src/traits.rs

@@ -0,0 +1,16 @@
+use crate::rate::Rate;
+
+pub trait LatestRate {
+    type Error: std::error::Error + Send + Sync + 'static;
+
+    fn latest_rate(&mut self) -> Result<Rate, Self::Error>;
+}
+
+// Future: Allow for different price feed sources
+pub trait PriceFeed: Sized {
+    type Error: std::error::Error + Send + Sync + 'static;
+    type Update;
+    
+    async fn connect(url: url::Url) -> Result<Self, Self::Error>;
+    async fn next_update(&mut self) -> Result<Self::Update, Self::Error>;
+}

swap/Cargo.toml

@@ -70,6 +70,7 @@ structopt = "0.3"
 strum = { version = "0.26", features = ["derive"] }
 swap-env = { path = "../swap-env" }
 swap-fs = { path = "../swap-fs" }
+swap-feed = { path = "../swap-feed" }
 swap-serde = { path = "../swap-serde" }
 tauri = { version = "2.0", features = ["config-json5"], optional = true, default-features = false }
 thiserror = { workspace = true }
@@ -82,7 +83,7 @@ tower = { version = "0.4.13", features = ["full"] }
 tower-http = { version = "0.3.4", features = ["full"] }
 tracing = { workspace = true }
 tracing-appender = "0.2"
-tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt", "ansi", "env-filter", "time", "tracing-log", "json"] }
+tracing-subscriber = { workspace = true }
 typeshare = { workspace = true }
 unsigned-varint = { version = "0.8.0", features = ["codec", "asynchronous_codec"] }
 url = { workspace = true }

swap/src/asb.rs

@@ -1,14 +1,13 @@
 pub mod command;
 mod event_loop;
 mod network;
-mod rate;
 mod recovery;
 
-pub use event_loop::{EventLoop, EventLoopHandle, FixedRate, KrakenRate, LatestRate};
+pub use event_loop::{EventLoop, EventLoopHandle};
 pub use network::behaviour::{Behaviour, OutEvent};
 pub use network::rendezvous::RendezvousNode;
 pub use network::transport;
-pub use rate::Rate;
+pub use swap_feed::{FixedRate, KrakenRate, LatestRate, Rate};
 pub use recovery::cancel::cancel;
 pub use recovery::punish::punish;
 pub use recovery::redeem::{redeem, Finality};

swap/src/asb/event_loop.rs

@@ -1,4 +1,4 @@
-use crate::asb::{Behaviour, OutEvent, Rate};
+use crate::asb::{Behaviour, OutEvent};
 use crate::network::cooperative_xmr_redeem_after_punish::CooperativeXmrRedeemRejectReason;
 use crate::network::cooperative_xmr_redeem_after_punish::Response::{Fullfilled, Rejected};
 use crate::network::quote::BidQuote;
@@ -7,7 +7,8 @@ use crate::network::transfer_proof;
 use crate::protocol::alice::swap::has_already_processed_enc_sig;
 use crate::protocol::alice::{AliceState, ReservesMonero, State3, Swap};
 use crate::protocol::{Database, State};
-use crate::{bitcoin, kraken, monero};
+use crate::{bitcoin, monero};
+use swap_feed::{LatestRate};
 use swap_env::env;
 use anyhow::{anyhow, Context, Result};
 use futures::future;
@@ -18,9 +19,8 @@ use libp2p::swarm::SwarmEvent;
 use libp2p::{PeerId, Swarm};
 use moka::future::Cache;
 use monero::Amount;
-use rust_decimal::Decimal;
 use std::collections::HashMap;
-use std::convert::{Infallible, TryInto};
+use std::convert::{TryInto};
 use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::Duration;
@@ -615,67 +615,6 @@ where
     }
 }
 
-pub trait LatestRate {
-    type Error: std::error::Error + Send + Sync + 'static;
-
-    fn latest_rate(&mut self) -> Result<Rate, Self::Error>;
-}
-
-#[derive(Clone, Debug)]
-pub struct FixedRate(Rate);
-
-impl FixedRate {
-    pub const RATE: f64 = 0.01;
-
-    pub fn value(&self) -> Rate {
-        self.0
-    }
-}
-
-impl Default for FixedRate {
-    fn default() -> Self {
-        let ask = bitcoin::Amount::from_btc(Self::RATE).expect("Static value should never fail");
-        let spread = Decimal::from(0u64);
-
-        Self(Rate::new(ask, spread))
-    }
-}
-
-impl LatestRate for FixedRate {
-    type Error = Infallible;
-
-    fn latest_rate(&mut self) -> Result<Rate, Self::Error> {
-        Ok(self.value())
-    }
-}
-
-/// Produces [`Rate`]s based on [`PriceUpdate`]s from kraken and a configured
-/// spread.
-#[derive(Debug, Clone)]
-pub struct KrakenRate {
-    ask_spread: Decimal,
-    price_updates: kraken::PriceUpdates,
-}
-
-impl KrakenRate {
-    pub fn new(ask_spread: Decimal, price_updates: kraken::PriceUpdates) -> Self {
-        Self {
-            ask_spread,
-            price_updates,
-        }
-    }
-}
-
-impl LatestRate for KrakenRate {
-    type Error = kraken::Error;
-
-    fn latest_rate(&mut self) -> Result<Rate, Self::Error> {
-        let update = self.price_updates.latest_update()?;
-        let rate = Rate::new(update.ask, self.ask_spread);
-
-        Ok(rate)
-    }
-}
 
 #[derive(Debug)]
 pub struct EventLoopHandle {

swap/src/asb/network.rs

@@ -1,4 +1,4 @@
-use crate::asb::event_loop::LatestRate;
+use swap_feed::LatestRate;
 use swap_env::env;
 use crate::network::quote::BidQuote;
 use crate::network::rendezvous::XmrBtcNamespace;

swap/src/bin/asb.rs

@@ -38,7 +38,8 @@ use swap::protocol::alice::swap::is_complete;
 use swap::protocol::alice::{run, AliceState};
 use swap::protocol::{Database, State};
 use swap::seed::Seed;
-use swap::{bitcoin, kraken, monero};
+use swap::{bitcoin, monero};
+use swap_feed;
 use tracing_subscriber::filter::LevelFilter;
 use uuid::Uuid;
 
@@ -194,7 +195,7 @@ pub async fn main() -> Result<()> {
             tracing::info!(%bitcoin_balance, "Bitcoin wallet balance");
 
             // Connect to Kraken
-            let kraken_price_updates = kraken::connect(config.maker.price_ticker_ws_url.clone())?;
+            let kraken_price_updates = swap_feed::connect_kraken(config.maker.price_ticker_ws_url.clone())?;
 
             let kraken_rate = KrakenRate::new(config.maker.ask_spread, kraken_price_updates);
             let namespace = XmrBtcNamespace::from_is_testnet(testnet);

swap/src/cli/cancel_and_refund.rs

@@ -60,7 +60,6 @@ pub async fn cancel(
         BobState::BtcEarlyRefundPublished(state6) => state6,
 
         BobState::Started { .. }
-        | BobState::SwapSetupCompleted(_)
         | BobState::BtcRedeemed(_)
         | BobState::XmrRedeemed { .. }
         | BobState::BtcPunished { .. }

swap/src/cli/command.rs

@@ -4,7 +4,6 @@ use crate::cli::api::request::{
     GetHistoryArgs, ListSellersArgs, MoneroRecoveryArgs, Request, ResumeSwapArgs, WithdrawBtcArgs,
 };
 use crate::cli::api::Context;
-use swap_serde::monero::address;
 use crate::monero::{self, MoneroAddressPool};
 use anyhow::Result;
 use bitcoin::address::NetworkUnchecked;

swap/src/lib.rs

@@ -21,7 +21,6 @@ pub mod bitcoin;
 pub mod cli;
 pub mod common;
 pub mod database;
-pub mod kraken;
 pub mod libp2p_ext;
 pub mod monero;
 mod monero_ext;

swap/src/network/swap_setup/alice.rs

@@ -381,8 +381,8 @@ where
 
                         let unlocked = wallet_snapshot.unlocked_balance;
 
-                        let needed_balance = xmr + wallet_snapshot.lock_fee;
-                        if unlocked.as_piconero() < needed_balance.as_piconero() {
+                        let needed_balance = xmr + wallet_snapshot.lock_fee.into();
+                        if unlocked.as_piconero() < needed_balance.as_pico() {
                             tracing::warn!(
                                 unlocked_balance = %unlocked,
                                 needed_balance = %needed_balance,
@@ -399,14 +399,18 @@ where
 
                     let result = validate.await;
 
+                    let converted_result = match result {
+                        Ok(xmr) => Ok(xmr.into()),
+                        Err(e) => Err(e),
+                    };
                     swap_setup::write_cbor_message(
                         &mut substream,
-                        SpotPriceResponse::from_result_ref(&result),
+                        SpotPriceResponse::from_result_ref(&converted_result),
                     )
                     .await
                     .context("Failed to write spot price response")?;
 
-                    let xmr = result?;
+                    let xmr = converted_result?;
 
                     let state0 = State0::new(
                         request.btc,