initiating swaps in a separate task and handling shutdown signals with broadcast queues

This commit is contained in:
Lorenzo Tucci 2023-08-14 11:31:54 +02:00
commit 969c58e987
12 changed files with 353 additions and 283 deletions

View file

@ -129,7 +129,7 @@ impl Context {
let tor_socks5_port = tor.map(|tor| tor.tor_socks5_port); let tor_socks5_port = tor.map(|tor| tor.tor_socks5_port);
START.call_once(|| { START.call_once(|| {
let _ = cli::tracing::init(debug, json, data_dir.join("logs"), None); let _ = cli::tracing::init(debug, json, data_dir.join("logs"));
}); });
let context = Context { let context = Context {

View file

@ -1,5 +1,5 @@
use crate::api::Context; use crate::api::Context;
use crate::bitcoin::{Amount, TxLock}; use crate::bitcoin::{Amount, ExpiredTimelocks, TxLock};
use crate::cli::{list_sellers, EventLoop, SellerStatus}; use crate::cli::{list_sellers, EventLoop, SellerStatus};
use crate::libp2p_ext::MultiAddrExt; use crate::libp2p_ext::MultiAddrExt;
use crate::network::quote::{BidQuote, ZeroQuoteReceived}; use crate::network::quote::{BidQuote, ZeroQuoteReceived};
@ -18,81 +18,90 @@ use std::future::Future;
use std::net::SocketAddr; use std::net::SocketAddr;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use structopt::lazy_static::lazy_static;
use tracing::{debug_span, Instrument}; use tracing::{debug_span, Instrument};
use uuid::Uuid; use uuid::Uuid;
use tokio::sync::RwLock;
lazy_static! {
static ref SWAP_LOCK: RwLock<Option<Uuid>> = RwLock::new(None);
}
#[derive(PartialEq, Debug)] #[derive(PartialEq, Debug)]
pub struct Request { pub struct Request {
pub params: Params,
pub cmd: Method pub cmd: Method
} }
#[derive(Default, PartialEq, Debug)]
pub struct Params {
pub seller: Option<Multiaddr>,
pub bitcoin_change_address: Option<bitcoin::Address>,
pub monero_receive_address: Option<monero::Address>,
pub rendezvous_point: Option<Multiaddr>,
pub swap_id: Option<Uuid>,
pub amount: Option<Amount>,
pub server_address: Option<SocketAddr>,
pub address: Option<bitcoin::Address>,
}
#[derive(Debug, PartialEq)] #[derive(Debug, PartialEq)]
pub enum Method { pub enum Method {
BuyXmr, BuyXmr {
seller: Multiaddr,
bitcoin_change_address: bitcoin::Address,
monero_receive_address: monero::Address,
swap_id: Uuid,
},
History, History,
RawHistory, RawHistory,
Config, Config,
WithdrawBtc, WithdrawBtc {
amount: Option<Amount>,
address: bitcoin::Address,
},
Balance, Balance,
GetSeller, GetSeller {
SwapStartDate, swap_id: Uuid,
Resume, },
CancelAndRefund, SwapStartDate {
ListSellers, swap_id: Uuid,
},
Resume {
swap_id: Uuid,
},
CancelAndRefund {
swap_id: Uuid,
},
ListSellers {
rendezvous_point: Multiaddr,
},
ExportBitcoinWallet, ExportBitcoinWallet,
MoneroRecovery, MoneroRecovery {
StartDaemon, swap_id: Uuid,
},
StartDaemon {
server_address: Option<SocketAddr>,
},
GetCurrentSwap,
GetSwapExpiredTimelock {
swap_id: Uuid,
},
} }
impl Request { impl Request {
pub fn new(cmd: Method, params: Params) -> Request { pub fn new(cmd: Method) -> Request {
Request { Request {
params,
cmd cmd
} }
} }
async fn handle_cmd(&mut self, context: Arc<Context>) -> Result<serde_json::Value> { fn has_lockable_swap_id(&self) -> Option<Uuid> {
match self.cmd { match self.cmd {
Method::BuyXmr => { Method::BuyXmr { swap_id, .. }
let swap_id = self | Method::Resume { swap_id }
.params | Method::CancelAndRefund { swap_id } => Some(swap_id),
.swap_id _ => None,
.context("Parameter swap_id is missing")?; }
}
async fn handle_cmd(self, context: Arc<Context>) -> Result<serde_json::Value> {
match self.cmd {
Method::BuyXmr { seller, bitcoin_change_address, monero_receive_address, swap_id } => {
let seed = context.config.seed.as_ref().context("Could not get seed")?; let seed = context.config.seed.as_ref().context("Could not get seed")?;
let env_config = context.config.env_config; let env_config = context.config.env_config;
let btc = context let btc = context
.bitcoin_wallet .bitcoin_wallet
.as_ref() .as_ref()
.context("Could not get Bitcoin wallet")?; .context("Could not get Bitcoin wallet")?;
let seller = self
.params
.seller
.clone()
.context("Parameter seller is missing")?;
let monero_receive_address = self
.params
.monero_receive_address
.context("Parameter monero_receive_address is missing")?;
let bitcoin_change_address = self
.params
.bitcoin_change_address
.clone()
.context("Parameter bitcoin_change_address is missing")?;
let bitcoin_wallet = btc; let bitcoin_wallet = btc;
let seller_peer_id = seller let seller_peer_id = seller
@ -224,8 +233,7 @@ impl Request {
Ok(json!({ "raw_history": raw_history })) Ok(json!({ "raw_history": raw_history }))
} }
Method::GetSeller => { Method::GetSeller { swap_id } => {
let swap_id = self.params.swap_id.context("Parameter swap_id is needed")?;
let peerId = context let peerId = context
.db .db
.get_peer_id(swap_id) .get_peer_id(swap_id)
@ -243,12 +251,7 @@ impl Request {
"addresses": addresses "addresses": addresses
})) }))
} }
Method::SwapStartDate => { Method::SwapStartDate { swap_id } => {
let swap_id = self
.params
.swap_id
.context("Parameter swap_id is missing")?;
let start_date = context.db.get_swap_start_date(swap_id).await?; let start_date = context.db.get_swap_start_date(swap_id).await?;
Ok(json!({ Ok(json!({
@ -272,19 +275,13 @@ impl Request {
"bitcoin_wallet": format!("{}/wallet", data_dir_display), "bitcoin_wallet": format!("{}/wallet", data_dir_display),
})) }))
} }
Method::WithdrawBtc => { Method::WithdrawBtc {address, amount} => {
let bitcoin_wallet = context let bitcoin_wallet = context
.bitcoin_wallet .bitcoin_wallet
.as_ref() .as_ref()
.context("Could not get Bitcoin wallet")?; .context("Could not get Bitcoin wallet")?;
let address = self let amount = match amount {
.params
.address
.clone()
.context("Parameter address is missing")?;
let amount = match self.params.amount {
Some(amount) => amount, Some(amount) => amount,
None => { None => {
bitcoin_wallet bitcoin_wallet
@ -307,12 +304,9 @@ impl Request {
"txid": signed_tx.txid(), "txid": signed_tx.txid(),
})) }))
} }
Method::StartDaemon => { Method::StartDaemon {server_address} => {
// Default to 127.0.0.1:1234 // Default to 127.0.0.1:1234
let server_address = self let server_address = server_address.unwrap_or("127.0.0.1:1234".parse().unwrap());
.params
.server_address
.unwrap_or("127.0.0.1:1234".parse().unwrap());
let (_, server_handle) = let (_, server_handle) =
rpc::run_server(server_address, Arc::clone(&context)).await?; rpc::run_server(server_address, Arc::clone(&context)).await?;
@ -347,12 +341,7 @@ impl Request {
"balance": bitcoin_balance.to_sat() "balance": bitcoin_balance.to_sat()
})) }))
} }
Method::Resume => { Method::Resume {swap_id} => {
let swap_id = self
.params
.swap_id
.context("Parameter swap_id is missing")?;
let seller_peer_id = context.db.get_peer_id(swap_id).await?; let seller_peer_id = context.db.get_peer_id(swap_id).await?;
let seller_addresses = context.db.get_addresses(seller_peer_id).await?; let seller_addresses = context.db.get_addresses(seller_peer_id).await?;
@ -431,16 +420,14 @@ impl Request {
"result": [] "result": []
})) }))
} }
Method::CancelAndRefund => { Method::CancelAndRefund {swap_id} => {
let bitcoin_wallet = context let bitcoin_wallet = context
.bitcoin_wallet .bitcoin_wallet
.as_ref() .as_ref()
.context("Could not get Bitcoin wallet")?; .context("Could not get Bitcoin wallet")?;
let state = cli::cancel_and_refund( let state = cli::cancel_and_refund(
self.params swap_id,
.swap_id
.context("Parameter swap_id is missing")?,
Arc::clone(bitcoin_wallet), Arc::clone(bitcoin_wallet),
Arc::clone(&context.db), Arc::clone(&context.db),
) )
@ -450,12 +437,7 @@ impl Request {
"result": state, "result": state,
})) }))
} }
Method::ListSellers => { Method::ListSellers {rendezvous_point} => {
let rendezvous_point = self
.params
.rendezvous_point
.clone()
.context("Parameter rendezvous_point is missing")?;
let rendezvous_node_peer_id = rendezvous_point let rendezvous_node_peer_id = rendezvous_point
.extract_peer_id() .extract_peer_id()
.context("Rendezvous node address must contain peer ID")?; .context("Rendezvous node address must contain peer ID")?;
@ -512,16 +494,14 @@ impl Request {
let wallet_export = bitcoin_wallet.wallet_export("cli").await?; let wallet_export = bitcoin_wallet.wallet_export("cli").await?;
tracing::info!(descriptor=%wallet_export.to_string(), "Exported bitcoin wallet"); tracing::info!(descriptor=%wallet_export.to_string(), "Exported bitcoin wallet");
Ok(json!({ Ok(json!({
"result": [] "descriptor": wallet_export.to_string(),
})) }))
} }
Method::MoneroRecovery => { Method::MoneroRecovery {swap_id} => {
let swap_state: BobState = context let swap_state: BobState = context
.db .db
.get_state( .get_state(
self.params swap_id,
.swap_id
.context("Parameter swap_id is missing")?,
) )
.await? .await?
.try_into()?; .try_into()?;
@ -560,28 +540,71 @@ impl Request {
Ok(json!({ Ok(json!({
"result": [] "result": []
})) }))
} },
Method::GetCurrentSwap => {
Ok(json!({
"swap_id": SWAP_LOCK.read().await.clone()
}))
},
Method::GetSwapExpiredTimelock { swap_id } => {
let swap_state: BobState = context
.db
.get_state(
swap_id,
)
.await?
.try_into()?;
let bitcoin_wallet = context.bitcoin_wallet.as_ref().context("Could not get Bitcoin wallet")?;
let timelock = match swap_state {
BobState::Started { .. }
| BobState::SafelyAborted
| BobState::SwapSetupCompleted(_) => bail!("Bitcoin lock transaction has not been published yet"),
BobState::BtcLocked { state3: state, .. }
| BobState::XmrLockProofReceived { state, .. } => state.expired_timelock(bitcoin_wallet).await,
BobState::XmrLocked(state)
| BobState::EncSigSent(state) => state.expired_timelock(bitcoin_wallet).await,
BobState::CancelTimelockExpired(state)
| BobState::BtcCancelled(state) => state.expired_timelock(bitcoin_wallet).await,
BobState::BtcPunished { .. } => Ok(ExpiredTimelocks::Punish),
// swap is already finished
BobState::BtcRefunded(_)
| BobState::BtcRedeemed(_)
| BobState::XmrRedeemed { .. } => bail!("Bitcoin have already been redeemed or refunded")
}?;
Ok(json!({
"timelock": timelock,
}))
},
} }
} }
pub async fn call(&mut self, context: Arc<Context>) -> Result<serde_json::Value> { pub async fn call(self, context: Arc<Context>) -> Result<serde_json::Value> {
// If the swap ID is set, we add it to the span // If the swap ID is set, we add it to the span
let call_span = self.params.swap_id.map_or_else( let call_span = debug_span!(
|| { "cmd",
debug_span!(
"call",
method = ?self.cmd, method = ?self.cmd,
)
},
|swap_id| {
debug_span!(
"call",
method = ?self.cmd,
swap_id = swap_id.to_string(),
)
},
); );
if let Some(swap_id) = self.has_lockable_swap_id() {
println!("taking lock for swap_id: {}", swap_id);
let mut guard = SWAP_LOCK.write().await;
if let Some(running_swap_id) = guard.as_ref() {
bail!("Another swap is already running: {}", running_swap_id);
}
let _ = guard.insert(swap_id.clone());
drop(guard);
let result = self.handle_cmd(context).instrument(call_span).await;
SWAP_LOCK.write().await.take();
println!("releasing lock for swap_id: {}", swap_id);
return result;
}
self.handle_cmd(context).instrument(call_span).await self.handle_cmd(context).instrument(call_span).await
} }
} }
@ -604,7 +627,7 @@ pub async fn determine_btc_to_swap<FB, TB, FMG, TMG, FS, TS, FFE, TFE>(
sync: FS, sync: FS,
estimate_fee: FFE, estimate_fee: FFE,
) -> Result<(bitcoin::Amount, bitcoin::Amount)> ) -> Result<(bitcoin::Amount, bitcoin::Amount)>
where where
TB: Future<Output = Result<bitcoin::Amount>>, TB: Future<Output = Result<bitcoin::Amount>>,
FB: Fn() -> TB, FB: Fn() -> TB,
TMG: Future<Output = Result<bitcoin::Amount>>, TMG: Future<Output = Result<bitcoin::Amount>>,
@ -687,3 +710,4 @@ where
Ok((btc_swap_amount, fees)) Ok((btc_swap_amount, fees))
} }

View file

@ -244,10 +244,14 @@ pub fn current_epoch(
} }
if tx_lock_status.is_confirmed_with(cancel_timelock) { if tx_lock_status.is_confirmed_with(cancel_timelock) {
return ExpiredTimelocks::Cancel; return ExpiredTimelocks::Cancel {
blocks_left: tx_cancel_status.blocks_left_until(punish_timelock),
}
} }
ExpiredTimelocks::None ExpiredTimelocks::None {
blocks_left: tx_lock_status.blocks_left_until(cancel_timelock),
}
} }
pub mod bitcoin_address { pub mod bitcoin_address {

View file

@ -24,6 +24,12 @@ use std::ops::Add;
#[serde(transparent)] #[serde(transparent)]
pub struct CancelTimelock(u32); pub struct CancelTimelock(u32);
impl From<CancelTimelock> for u32 {
fn from(cancel_timelock: CancelTimelock) -> Self {
cancel_timelock.0
}
}
impl CancelTimelock { impl CancelTimelock {
pub const fn new(number_of_blocks: u32) -> Self { pub const fn new(number_of_blocks: u32) -> Self {
Self(number_of_blocks) Self(number_of_blocks)
@ -64,6 +70,12 @@ impl fmt::Display for CancelTimelock {
#[serde(transparent)] #[serde(transparent)]
pub struct PunishTimelock(u32); pub struct PunishTimelock(u32);
impl From<PunishTimelock> for u32 {
fn from(punish_timelock: PunishTimelock) -> Self {
punish_timelock.0
}
}
impl PunishTimelock { impl PunishTimelock {
pub const fn new(number_of_blocks: u32) -> Self { pub const fn new(number_of_blocks: u32) -> Self {
Self(number_of_blocks) Self(number_of_blocks)

View file

@ -37,9 +37,13 @@ impl Add<u32> for BlockHeight {
} }
} }
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Serialize, Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExpiredTimelocks { pub enum ExpiredTimelocks {
None, None {
Cancel, blocks_left: u32,
},
Cancel {
blocks_left: u32,
},
Punish, Punish,
} }

View file

@ -274,7 +274,7 @@ impl Subscription {
pub async fn wait_until_confirmed_with<T>(&self, target: T) -> Result<()> pub async fn wait_until_confirmed_with<T>(&self, target: T) -> Result<()>
where where
u32: PartialOrd<T>, T: Into<u32>,
T: Copy, T: Copy,
{ {
self.wait_until(|status| status.is_confirmed_with(target)) self.wait_until(|status| status.is_confirmed_with(target))
@ -926,10 +926,19 @@ impl Confirmed {
} }
pub fn meets_target<T>(&self, target: T) -> bool pub fn meets_target<T>(&self, target: T) -> bool
where where T: Into<u32>
u32: PartialOrd<T>,
{ {
self.confirmations() >= target self.confirmations() >= target.into()
}
pub fn blocks_left_until<T>(&self, target: T) -> u32
where T: Into<u32>, T: Copy
{
if self.meets_target(target) {
0
} else {
target.into() - self.confirmations()
}
} }
} }
@ -941,8 +950,7 @@ impl ScriptStatus {
/// Check if the script has met the given confirmation target. /// Check if the script has met the given confirmation target.
pub fn is_confirmed_with<T>(&self, target: T) -> bool pub fn is_confirmed_with<T>(&self, target: T) -> bool
where where T: Into<u32>
u32: PartialOrd<T>,
{ {
match self { match self {
ScriptStatus::Confirmed(inner) => inner.meets_target(target), ScriptStatus::Confirmed(inner) => inner.meets_target(target),
@ -950,6 +958,18 @@ impl ScriptStatus {
} }
} }
// Calculate the number of blocks left until the target is met.
pub fn blocks_left_until<T>(&self, target: T) -> u32
where T: Into<u32>, T: Copy
{
match self {
ScriptStatus::Confirmed(inner) => {
inner.blocks_left_until(target)
}
_ => target.into(),
}
}
pub fn has_been_seen(&self) -> bool { pub fn has_been_seen(&self) -> bool {
matches!(self, ScriptStatus::InMempool | ScriptStatus::Confirmed(_)) matches!(self, ScriptStatus::InMempool | ScriptStatus::Confirmed(_))
} }
@ -1005,6 +1025,33 @@ mod tests {
assert_eq!(confirmed.depth, 0) assert_eq!(confirmed.depth, 0)
} }
#[test]
fn given_depth_0_should_return_0_blocks_left_until_1() {
let script = ScriptStatus::Confirmed(Confirmed { depth: 0 });
let blocks_left = script.blocks_left_until(1);
assert_eq!(blocks_left, 0)
}
#[test]
fn given_depth_1_should_return_0_blocks_left_until_1() {
let script = ScriptStatus::Confirmed(Confirmed { depth: 1 });
let blocks_left = script.blocks_left_until(1);
assert_eq!(blocks_left, 0)
}
#[test]
fn given_depth_0_should_return_1_blocks_left_until_2() {
let script = ScriptStatus::Confirmed(Confirmed { depth: 0 });
let blocks_left = script.blocks_left_until(2);
assert_eq!(blocks_left, 1)
}
#[test] #[test]
fn given_one_BTC_and_100k_sats_per_vb_fees_should_not_hit_max() { fn given_one_BTC_and_100k_sats_per_vb_fees_should_not_hit_max() {
// 400 weight = 100 vbyte // 400 weight = 100 vbyte

View file

@ -1,4 +1,4 @@
use crate::api::request::{Method, Params, Request}; use crate::api::request::{Method, Request};
use crate::api::Context; use crate::api::Context;
use crate::bitcoin::{bitcoin_address, Amount}; use crate::bitcoin::{bitcoin_address, Amount};
use crate::monero; use crate::monero;
@ -79,13 +79,12 @@ where
bitcoin_address::validate(bitcoin_change_address, is_testnet)?; bitcoin_address::validate(bitcoin_change_address, is_testnet)?;
let request = Request::new( let request = Request::new(
Method::BuyXmr, Method::BuyXmr {
Params { seller,
bitcoin_change_address: Some(bitcoin_change_address), bitcoin_change_address,
monero_receive_address: Some(monero_receive_address), monero_receive_address,
seller: Some(seller), swap_id: Uuid::new_v4(),
..Default::default() }
},
); );
let context = Context::build( let context = Context::build(
@ -103,21 +102,21 @@ where
(context, request) (context, request)
} }
CliCommand::History => { CliCommand::History => {
let request = Request::new(Method::History, Params::default()); let request = Request::new(Method::History);
let context = let context =
Context::build(None, None, None, data, is_testnet, debug, json, None, rx).await?; Context::build(None, None, None, data, is_testnet, debug, json, None, rx).await?;
(context, request) (context, request)
} }
CliCommand::Config => { CliCommand::Config => {
let request = Request::new(Method::Config, Params::default()); let request = Request::new(Method::Config);
let context = let context =
Context::build(None, None, None, data, is_testnet, debug, json, None, rx).await?; Context::build(None, None, None, data, is_testnet, debug, json, None, rx).await?;
(context, request) (context, request)
} }
CliCommand::Balance { bitcoin } => { CliCommand::Balance { bitcoin } => {
let request = Request::new(Method::Balance, Params::default()); let request = Request::new(Method::Balance);
let context = Context::build( let context = Context::build(
Some(bitcoin), Some(bitcoin),
@ -140,11 +139,9 @@ where
tor, tor,
} => { } => {
let request = Request::new( let request = Request::new(
Method::StartDaemon, Method::StartDaemon {
Params { server_address
server_address, }
..Default::default()
},
); );
let context = Context::build( let context = Context::build(
@ -169,11 +166,9 @@ where
let address = bitcoin_address::validate(address, is_testnet)?; let address = bitcoin_address::validate(address, is_testnet)?;
let request = Request::new( let request = Request::new(
Method::WithdrawBtc, Method::WithdrawBtc {
Params {
amount, amount,
address: Some(address), address,
..Default::default()
}, },
); );
@ -198,11 +193,9 @@ where
tor, tor,
} => { } => {
let request = Request::new( let request = Request::new(
Method::Resume, Method::Resume {
Params { swap_id
swap_id: Some(swap_id), }
..Default::default()
},
); );
let context = Context::build( let context = Context::build(
@ -225,11 +218,9 @@ where
tor, tor,
} => { } => {
let request = Request::new( let request = Request::new(
Method::CancelAndRefund, Method::CancelAndRefund {
Params { swap_id
swap_id: Some(swap_id), }
..Default::default()
},
); );
let context = Context::build( let context = Context::build(
@ -251,11 +242,9 @@ where
tor, tor,
} => { } => {
let request = Request::new( let request = Request::new(
Method::ListSellers, Method::ListSellers {
Params { rendezvous_point
rendezvous_point: Some(rendezvous_point), }
..Default::default()
},
); );
let context = Context::build( let context = Context::build(
@ -276,7 +265,6 @@ where
CliCommand::ExportBitcoinWallet { bitcoin } => { CliCommand::ExportBitcoinWallet { bitcoin } => {
let request = Request::new( let request = Request::new(
Method::ExportBitcoinWallet, Method::ExportBitcoinWallet,
Params::default(),
); );
let context = Context::build( let context = Context::build(
@ -297,11 +285,9 @@ where
swap_id: SwapId { swap_id }, swap_id: SwapId { swap_id },
} => { } => {
let request = Request::new( let request = Request::new(
Method::MoneroRecovery, Method::MoneroRecovery {
Params { swap_id
swap_id: Some(swap_id), }
..Default::default()
},
); );
let context = let context =

View file

@ -7,17 +7,14 @@ use tracing::{Event, Level, Subscriber};
use tracing_subscriber::fmt::format::{DefaultFields, Format, JsonFields}; use tracing_subscriber::fmt::format::{DefaultFields, Format, JsonFields};
use tracing_subscriber::fmt::time::UtcTime; use tracing_subscriber::fmt::time::UtcTime;
use tracing_subscriber::layer::{Context, SubscriberExt}; use tracing_subscriber::layer::{Context, SubscriberExt};
use tracing_subscriber::{fmt, EnvFilter, FmtSubscriber, Layer, Registry}; use tracing_subscriber::{fmt, EnvFilter, Layer, Registry};
use uuid::Uuid;
pub fn init(debug: bool, json: bool, dir: impl AsRef<Path>, swap_id: Option<Uuid>) -> Result<()> { pub fn init(debug: bool, json: bool, dir: impl AsRef<Path>) -> Result<()> {
if let Some(swap_id) = swap_id {
let level_filter = EnvFilter::try_new("swap=debug")?; let level_filter = EnvFilter::try_new("swap=debug")?;
let registry = Registry::default().with(level_filter); let registry = Registry::default().with(level_filter);
let appender = let appender =
tracing_appender::rolling::never(dir.as_ref(), format!("swap-{}.log", swap_id)); tracing_appender::rolling::never(dir.as_ref(), "swap-all.log");
let (appender, guard) = tracing_appender::non_blocking(appender); let (appender, guard) = tracing_appender::non_blocking(appender);
std::mem::forget(guard); std::mem::forget(guard);
@ -26,6 +23,7 @@ pub fn init(debug: bool, json: bool, dir: impl AsRef<Path>, swap_id: Option<Uuid
fmt::layer() fmt::layer()
.with_ansi(false) .with_ansi(false)
.with_target(false) .with_target(false)
.with_span_events(fmt::format::FmtSpan::FULL)
.json() .json()
.with_writer(appender), .with_writer(appender),
); );
@ -39,23 +37,6 @@ pub fn init(debug: bool, json: bool, dir: impl AsRef<Path>, swap_id: Option<Uuid
} else { } else {
set_global_default(file_logger.with(info_terminal_printer()))?; set_global_default(file_logger.with(info_terminal_printer()))?;
} }
} else {
let level = if debug { Level::DEBUG } else { Level::INFO };
let is_terminal = atty::is(atty::Stream::Stderr);
let builder = FmtSubscriber::builder()
.with_env_filter(format!("swap={}", level))
.with_writer(std::io::stderr)
.with_ansi(is_terminal)
.with_timer(UtcTime::rfc_3339())
.with_target(false);
if json {
builder.json().init();
} else {
builder.init();
}
};
tracing::info!("Logging initialized to {}", dir.as_ref().display()); tracing::info!("Logging initialized to {}", dir.as_ref().display());
Ok(()) Ok(())
@ -66,17 +47,17 @@ pub struct StdErrPrinter<L> {
level: Level, level: Level,
} }
type StdErrLayer<S, T> = tracing_subscriber::fmt::Layer< type StdErrLayer<S, T> = fmt::Layer<
S, S,
DefaultFields, DefaultFields,
Format<tracing_subscriber::fmt::format::Full, T>, Format<fmt::format::Full, T>,
fn() -> std::io::Stderr, fn() -> std::io::Stderr,
>; >;
type StdErrJsonLayer<S, T> = tracing_subscriber::fmt::Layer< type StdErrJsonLayer<S, T> = fmt::Layer<
S, S,
JsonFields, JsonFields,
Format<tracing_subscriber::fmt::format::Json, T>, Format<fmt::format::Json, T>,
fn() -> std::io::Stderr, fn() -> std::io::Stderr,
>; >;

View file

@ -112,7 +112,7 @@ where
} }
AliceState::BtcLocked { state3 } => { AliceState::BtcLocked { state3 } => {
match state3.expired_timelocks(bitcoin_wallet).await? { match state3.expired_timelocks(bitcoin_wallet).await? {
ExpiredTimelocks::None => { ExpiredTimelocks::None {..} => {
// Record the current monero wallet block height so we don't have to scan from // Record the current monero wallet block height so we don't have to scan from
// block 0 for scenarios where we create a refund wallet. // block 0 for scenarios where we create a refund wallet.
let monero_wallet_restore_blockheight = monero_wallet.block_height().await?; let monero_wallet_restore_blockheight = monero_wallet.block_height().await?;
@ -135,7 +135,7 @@ where
transfer_proof, transfer_proof,
state3, state3,
} => match state3.expired_timelocks(bitcoin_wallet).await? { } => match state3.expired_timelocks(bitcoin_wallet).await? {
ExpiredTimelocks::None => { ExpiredTimelocks::None {..} => {
monero_wallet monero_wallet
.watch_for_transfer(state3.lock_xmr_watch_request(transfer_proof.clone(), 1)) .watch_for_transfer(state3.lock_xmr_watch_request(transfer_proof.clone(), 1))
.await .await
@ -221,7 +221,7 @@ where
encrypted_signature, encrypted_signature,
state3, state3,
} => match state3.expired_timelocks(bitcoin_wallet).await? { } => match state3.expired_timelocks(bitcoin_wallet).await? {
ExpiredTimelocks::None => { ExpiredTimelocks::None {..} => {
let tx_lock_status = bitcoin_wallet.subscribe_to(state3.tx_lock.clone()).await; let tx_lock_status = bitcoin_wallet.subscribe_to(state3.tx_lock.clone()).await;
match state3.signed_redeem_transaction(*encrypted_signature) { match state3.signed_redeem_transaction(*encrypted_signature) {
Ok(tx) => match bitcoin_wallet.broadcast(tx, "redeem").await { Ok(tx) => match bitcoin_wallet.broadcast(tx, "redeem").await {

View file

@ -440,7 +440,7 @@ impl State3 {
self.tx_lock.txid() self.tx_lock.txid()
} }
pub async fn current_epoch( pub async fn expired_timelock(
&self, &self,
bitcoin_wallet: &bitcoin::Wallet, bitcoin_wallet: &bitcoin::Wallet,
) -> Result<ExpiredTimelocks> { ) -> Result<ExpiredTimelocks> {

View file

@ -117,7 +117,7 @@ async fn next_state(
} => { } => {
let tx_lock_status = bitcoin_wallet.subscribe_to(state3.tx_lock.clone()).await; let tx_lock_status = bitcoin_wallet.subscribe_to(state3.tx_lock.clone()).await;
if let ExpiredTimelocks::None = state3.current_epoch(bitcoin_wallet).await? { if let ExpiredTimelocks::None {..} = state3.expired_timelock(bitcoin_wallet).await? {
let transfer_proof_watcher = event_loop_handle.recv_transfer_proof(); let transfer_proof_watcher = event_loop_handle.recv_transfer_proof();
let cancel_timelock_expires = let cancel_timelock_expires =
tx_lock_status.wait_until_confirmed_with(state3.cancel_timelock); tx_lock_status.wait_until_confirmed_with(state3.cancel_timelock);
@ -156,7 +156,7 @@ async fn next_state(
} => { } => {
let tx_lock_status = bitcoin_wallet.subscribe_to(state.tx_lock.clone()).await; let tx_lock_status = bitcoin_wallet.subscribe_to(state.tx_lock.clone()).await;
if let ExpiredTimelocks::None = state.current_epoch(bitcoin_wallet).await? { if let ExpiredTimelocks::None {..} = state.expired_timelock(bitcoin_wallet).await? {
let watch_request = state.lock_xmr_watch_request(lock_transfer_proof); let watch_request = state.lock_xmr_watch_request(lock_transfer_proof);
select! { select! {
@ -185,7 +185,7 @@ async fn next_state(
BobState::XmrLocked(state) => { BobState::XmrLocked(state) => {
let tx_lock_status = bitcoin_wallet.subscribe_to(state.tx_lock.clone()).await; let tx_lock_status = bitcoin_wallet.subscribe_to(state.tx_lock.clone()).await;
if let ExpiredTimelocks::None = state.expired_timelock(bitcoin_wallet).await? { if let ExpiredTimelocks::None {..} = state.expired_timelock(bitcoin_wallet).await? {
// Alice has locked Xmr // Alice has locked Xmr
// Bob sends Alice his key // Bob sends Alice his key
@ -209,7 +209,7 @@ async fn next_state(
BobState::EncSigSent(state) => { BobState::EncSigSent(state) => {
let tx_lock_status = bitcoin_wallet.subscribe_to(state.tx_lock.clone()).await; let tx_lock_status = bitcoin_wallet.subscribe_to(state.tx_lock.clone()).await;
if let ExpiredTimelocks::None = state.expired_timelock(bitcoin_wallet).await? { if let ExpiredTimelocks::None {..} = state.expired_timelock(bitcoin_wallet).await? {
select! { select! {
state5 = state.watch_for_redeem_btc(bitcoin_wallet) => { state5 = state.watch_for_redeem_btc(bitcoin_wallet) => {
BobState::BtcRedeemed(state5?) BobState::BtcRedeemed(state5?)
@ -269,12 +269,12 @@ async fn next_state(
BobState::BtcCancelled(state) => { BobState::BtcCancelled(state) => {
// Bob has cancelled the swap // Bob has cancelled the swap
match state.expired_timelock(bitcoin_wallet).await? { match state.expired_timelock(bitcoin_wallet).await? {
ExpiredTimelocks::None => { ExpiredTimelocks::None {..} => {
bail!( bail!(
"Internal error: canceled state reached before cancel timelock was expired" "Internal error: canceled state reached before cancel timelock was expired"
); );
} }
ExpiredTimelocks::Cancel => { ExpiredTimelocks::Cancel { .. } => {
state.publish_refund_btc(bitcoin_wallet).await?; state.publish_refund_btc(bitcoin_wallet).await?;
BobState::BtcRefunded(state) BobState::BtcRefunded(state)
} }

View file

@ -1,4 +1,4 @@
use crate::api::request::{Method, Params, Request}; use crate::api::request::{Method, Request};
use crate::api::Context; use crate::api::Context;
use crate::bitcoin::bitcoin_address; use crate::bitcoin::bitcoin_address;
use crate::monero::monero_address; use crate::monero::monero_address;
@ -18,19 +18,19 @@ pub fn register_modules(context: Arc<Context>) -> RpcModule<Arc<Context>> {
.register_async_method("get_bitcoin_balance", |_, context| async move { .register_async_method("get_bitcoin_balance", |_, context| async move {
get_bitcoin_balance(&context).await get_bitcoin_balance(&context).await
}) })
.expect("Could not register RPC method get_bitcoin_balance"); .unwrap();
module module
.register_async_method("get_history", |_, context| async move { .register_async_method("get_history", |_, context| async move {
get_history(&context).await get_history(&context).await
}) })
.expect("Could not register RPC method get_history"); .unwrap();
module module
.register_async_method("get_raw_history", |_, context| async move { .register_async_method("get_raw_history", |_, context| async move {
get_raw_history(&context).await get_raw_history(&context).await
}) })
.expect("Could not register RPC method get_raw_history"); .unwrap();
module module
.register_async_method("get_seller", |params, context| async move { .register_async_method("get_seller", |params, context| async move {
@ -42,7 +42,7 @@ pub fn register_modules(context: Arc<Context>) -> RpcModule<Arc<Context>> {
get_seller(*swap_id, &context).await get_seller(*swap_id, &context).await
}) })
.expect("Could not register RPC method get_seller"); .unwrap();
module module
.register_async_method("get_swap_start_date", |params, context| async move { .register_async_method("get_swap_start_date", |params, context| async move {
@ -54,7 +54,7 @@ pub fn register_modules(context: Arc<Context>) -> RpcModule<Arc<Context>> {
get_swap_start_date(*swap_id, &context).await get_swap_start_date(*swap_id, &context).await
}) })
.expect("Could not register RPC method get_swap_start_date"); .unwrap();
module module
.register_async_method("resume_swap", |params, context| async move { .register_async_method("resume_swap", |params, context| async move {
@ -66,7 +66,18 @@ pub fn register_modules(context: Arc<Context>) -> RpcModule<Arc<Context>> {
resume_swap(*swap_id, &context).await resume_swap(*swap_id, &context).await
}) })
.expect("Could not register RPC method resume_swap"); .unwrap();
module.register_async_method("get_swap_expired_timelock", |params, context| async move {
let params: HashMap<String, Uuid> = params.parse()?;
let swap_id = params.get("swap_id").ok_or_else(|| {
jsonrpsee_core::Error::Custom("Does not contain swap_id".to_string())
})?;
get_swap_timelock(*swap_id, &context).await
}).unwrap();
module module
.register_async_method("cancel_refund_swap", |params, context| async move { .register_async_method("cancel_refund_swap", |params, context| async move {
let params: HashMap<String, Uuid> = params.parse()?; let params: HashMap<String, Uuid> = params.parse()?;
@ -77,7 +88,7 @@ pub fn register_modules(context: Arc<Context>) -> RpcModule<Arc<Context>> {
cancel_and_refund_swap(*swap_id, &context).await cancel_and_refund_swap(*swap_id, &context).await
}) })
.expect("Could not register RPC method cancel_refund_swap"); .unwrap();
module module
.register_async_method("withdraw_btc", |params, context| async move { .register_async_method("withdraw_btc", |params, context| async move {
let params: HashMap<String, String> = params.parse()?; let params: HashMap<String, String> = params.parse()?;
@ -145,7 +156,7 @@ pub fn register_modules(context: Arc<Context>) -> RpcModule<Arc<Context>> {
) )
.await .await
}) })
.expect("Could not register RPC method buy_xmr"); .unwrap();
module module
.register_async_method("list_sellers", |params, context| async move { .register_async_method("list_sellers", |params, context| async move {
let params: HashMap<String, Multiaddr> = params.parse()?; let params: HashMap<String, Multiaddr> = params.parse()?;
@ -155,93 +166,99 @@ pub fn register_modules(context: Arc<Context>) -> RpcModule<Arc<Context>> {
list_sellers(rendezvous_point.clone(), &context).await list_sellers(rendezvous_point.clone(), &context).await
}) })
.expect("Could not register RPC method list_sellers"); .unwrap();
module.register_async_method("get_current_swap", |_, context| async move {
get_current_swap(&context).await
}).unwrap();
module module
} }
async fn execute_request( async fn execute_request(
cmd: Method, cmd: Method,
params: Params,
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
let mut request = Request::new(cmd, params); let request = Request::new(cmd);
request request
.call(Arc::clone(context)) .call(Arc::clone(context))
.await .await
.map_err(|err| jsonrpsee_core::Error::Custom(err.to_string())) .map_err(|err| jsonrpsee_core::Error::Custom(err.to_string()))
} }
async fn get_current_swap(context: &Arc<Context>) -> Result<serde_json::Value, jsonrpsee_core::Error> {
execute_request(Method::GetCurrentSwap, context).await
}
async fn get_bitcoin_balance( async fn get_bitcoin_balance(
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
execute_request(Method::Balance, Params::default(), context).await execute_request(Method::Balance, context).await
} }
async fn get_history(context: &Arc<Context>) -> Result<serde_json::Value, jsonrpsee_core::Error> { async fn get_history(context: &Arc<Context>) -> Result<serde_json::Value, jsonrpsee_core::Error> {
execute_request(Method::History, Params::default(), context).await execute_request(Method::History, context).await
} }
async fn get_raw_history( async fn get_raw_history(
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
execute_request(Method::RawHistory, Params::default(), context).await execute_request(Method::RawHistory, context).await
} }
async fn get_seller( async fn get_seller(
swap_id: Uuid, swap_id: Uuid,
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
let params = Params { execute_request(Method::GetSeller {
swap_id: Some(swap_id), swap_id
..Default::default() }, context).await
};
execute_request(Method::GetSeller, params, context).await
} }
async fn get_swap_start_date( async fn get_swap_start_date(
swap_id: Uuid, swap_id: Uuid,
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
let params = Params { execute_request(Method::SwapStartDate {
swap_id: Some(swap_id), swap_id
..Default::default() }, context).await
};
execute_request(Method::SwapStartDate, params, context).await
} }
async fn resume_swap( async fn resume_swap(
swap_id: Uuid, swap_id: Uuid,
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
let params = Params { execute_request(Method::Resume {
swap_id: Some(swap_id), swap_id
..Default::default() }, context).await
}; }
execute_request(Method::Resume, params, context).await
async fn get_swap_timelock(
swap_id: Uuid,
context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> {
execute_request(Method::GetSwapExpiredTimelock {
swap_id
}, context).await
} }
async fn cancel_and_refund_swap( async fn cancel_and_refund_swap(
swap_id: Uuid, swap_id: Uuid,
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
let params = Params { execute_request(Method::CancelAndRefund {
swap_id: Some(swap_id), swap_id
..Default::default() }, context).await
};
execute_request(Method::CancelAndRefund, params, context).await
} }
async fn withdraw_btc( async fn withdraw_btc(
withdraw_address: bitcoin::Address, address: bitcoin::Address,
amount: Option<bitcoin::Amount>, amount: Option<bitcoin::Amount>,
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
let params = Params { execute_request(Method::WithdrawBtc {
amount, amount,
address: Some(withdraw_address), address,
..Default::default() }, context).await
};
execute_request(Method::WithdrawBtc, params, context).await
} }
async fn buy_xmr( async fn buy_xmr(
@ -250,24 +267,19 @@ async fn buy_xmr(
seller: Multiaddr, seller: Multiaddr,
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
let params = Params { execute_request(Method::BuyXmr {
bitcoin_change_address: Some(bitcoin_change_address), seller,
monero_receive_address: Some(monero_receive_address), swap_id: Uuid::new_v4(),
seller: Some(seller), bitcoin_change_address,
swap_id: Some(Uuid::new_v4()), monero_receive_address
..Default::default() }, context).await
};
execute_request(Method::BuyXmr, params, context).await
} }
async fn list_sellers( async fn list_sellers(
rendezvous_point: Multiaddr, rendezvous_point: Multiaddr,
context: &Arc<Context>, context: &Arc<Context>,
) -> Result<serde_json::Value, jsonrpsee_core::Error> { ) -> Result<serde_json::Value, jsonrpsee_core::Error> {
let params = Params { execute_request(Method::ListSellers {
rendezvous_point: Some(rendezvous_point), rendezvous_point
..Default::default() }, context).await
};
execute_request(Method::ListSellers, params, context).await
} }