lukechilds
/
itchysats
mirror of https://github.com/lukechilds/itchysats.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Phase out log_error macro 

In general, we would like to let the sender of an `xtra::Message` to
be able to handle all the possible errors. The `log_error` macro
catered to the complete opposite use case, so we remove it.

In order to make error handling at the sending site more ergonomic, we
introduce a `LogFailure` trait designed for fallible `xtra::Messages`.

With this patch we've tried to change _where_ we handle errors, but
not _what_ we do with them.
update-blockstream-electrum-server-url
Lucas Soriano del Pino 3 years ago
parent
057d4628d1
commit
cf004aa84e
No known key found for this signature in database GPG Key ID: 89CE0DB40A19D524
15 changed files with 267 additions and 165 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 9
      daemon/src/actors.rs
  2. 7
      daemon/src/collab_settlement_maker.rs
  3. 13
      daemon/src/connection.rs
  4. 2
      daemon/src/lib.rs
  5. 247
      daemon/src/maker_cfd.rs
  6. 9
      daemon/src/maker_inc_connections.rs
  7. 19
      daemon/src/monitor.rs
  8. 14
      daemon/src/oracle.rs
  9. 6
      daemon/src/rollover_maker.rs
  10. 7
      daemon/src/setup_maker.rs
  11. 13
      daemon/src/setup_taker.rs
  12. 21
      daemon/src/taker_cfd.rs
  13. 50
      daemon/src/xtra_ext.rs
  14. 5
      daemon/tests/harness/mocks/monitor.rs
  15. 10
      daemon/tests/harness/mod.rs

9
daemon/src/actors.rs
View File

@ -1,9 +0,0 @@
/// Wrapper for handlers to log errors
#[macro_export]
macro_rules! log_error {
($future:expr) => {
if let Err(e) = $future.await {
tracing::error!("Message handler failed: {:#}", e);
}
};
}

7
daemon/src/collab_settlement_maker.rs
View File

@ -9,6 +9,7 @@ use crate::model::cfd::SettlementKind;
use crate::model::cfd::SettlementProposal; use crate::model::cfd::SettlementProposal;
use crate::model::Identity; use crate::model::Identity;
use crate::projection; use crate::projection;
use crate::xtra_ext::LogFailure;
use anyhow::Context; use anyhow::Context;
use async_trait::async_trait; use async_trait::async_trait;
use bdk::bitcoin::Script; use bdk::bitcoin::Script;
@ -172,7 +173,11 @@ impl Actor {
} }
async fn complete(&mut self, completed: Completed, ctx: &mut xtra::Context<Self>) { async fn complete(&mut self, completed: Completed, ctx: &mut xtra::Context<Self>) {
let _ = self.on_completed.send(completed).await; let _ = self
.on_completed
.send(completed)
.log_failure("Failed to inform about collab settlement completion")
.await;
ctx.stop(); ctx.stop();
} }

13
daemon/src/connection.rs
View File

@ -1,7 +1,6 @@
use crate::address_map::AddressMap; use crate::address_map::AddressMap;
use crate::address_map::Stopping; use crate::address_map::Stopping;
use crate::collab_settlement_taker; use crate::collab_settlement_taker;
use crate::log_error;
use crate::model::cfd::OrderId; use crate::model::cfd::OrderId;
use crate::model::Identity; use crate::model::Identity;
use crate::model::Price; use crate::model::Price;
@ -17,6 +16,7 @@ use crate::wire;
use crate::wire::EncryptedJsonCodec; use crate::wire::EncryptedJsonCodec;
use crate::wire::TakerToMaker; use crate::wire::TakerToMaker;
use crate::wire::Version; use crate::wire::Version;
use crate::xtra_ext::LogFailure;
use crate::Tasks; use crate::Tasks;
use anyhow::bail; use anyhow::bail;
use anyhow::Context; use anyhow::Context;
@ -146,7 +146,10 @@ impl Actor {
#[xtra_productivity(message_impl = false)] #[xtra_productivity(message_impl = false)]
impl Actor { impl Actor {
async fn handle_taker_to_maker(&mut self, message: wire::TakerToMaker) { async fn handle_taker_to_maker(&mut self, message: wire::TakerToMaker) {
log_error!(self.send_to_maker.send(message)); let msg_str = message.to_string();
if self.send_to_maker.send(message).await.is_err() {
tracing::warn!("Failed to send wire message {} to maker", msg_str);
}
} }
async fn handle_collab_settlement_actor_stopping( async fn handle_collab_settlement_actor_stopping(
@ -430,7 +433,11 @@ impl Actor {
} }
} }
wire::MakerToTaker::CurrentOrder(msg) => { wire::MakerToTaker::CurrentOrder(msg) => {
log_error!(self.current_order.send(CurrentOrder(msg))); let _ = self
.current_order
.send(CurrentOrder(msg))
.log_failure("Failed to forward current order from maker")
.await;
} }
wire::MakerToTaker::Hello(_) => { wire::MakerToTaker::Hello(_) => {
tracing::warn!("Ignoring unexpected Hello message from maker. Hello is only expected when opening a new connection.") tracing::warn!("Ignoring unexpected Hello message from maker. Hello is only expected when opening a new connection.")

2
daemon/src/lib.rs
View File

@ -26,7 +26,6 @@ use xtra::Address;
pub mod sqlx_ext; // Must come first because it is a macro. pub mod sqlx_ext; // Must come first because it is a macro.
pub mod actors;
pub mod address_map; pub mod address_map;
pub mod auth; pub mod auth;
pub mod auto_rollover; pub mod auto_rollover;
@ -67,6 +66,7 @@ pub mod try_continue;
pub mod tx; pub mod tx;
pub mod wallet; pub mod wallet;
pub mod wire; pub mod wire;
pub mod xtra_ext;
// Certain operations (e.g. contract setup) take long time in debug mode, // Certain operations (e.g. contract setup) take long time in debug mode,
// causing us to lag behind in processing heartbeats. // causing us to lag behind in processing heartbeats.

247
daemon/src/maker_cfd.rs
View File

@ -5,7 +5,6 @@ use crate::cfd_actors::insert_cfd_and_update_feed;
use crate::cfd_actors::{self}; use crate::cfd_actors::{self};
use crate::collab_settlement_maker; use crate::collab_settlement_maker;
use crate::db::load_cfd; use crate::db::load_cfd;
use crate::log_error;
use crate::maker_inc_connections; use crate::maker_inc_connections;
use crate::model::cfd::Cfd; use crate::model::cfd::Cfd;
use crate::model::cfd::CfdState; use crate::model::cfd::CfdState;
@ -557,49 +556,50 @@ where
M: xtra::Handler<monitor::CollaborativeSettlement>, M: xtra::Handler<monitor::CollaborativeSettlement>,
W: xtra::Handler<wallet::TryBroadcastTransaction>, W: xtra::Handler<wallet::TryBroadcastTransaction>,
{ {
async fn handle_settlement_completed(&mut self, msg: collab_settlement_maker::Completed) { async fn handle_settlement_completed(
log_error!(async { &mut self,
use collab_settlement_maker::Completed::*; msg: collab_settlement_maker::Completed,
let (order_id, settlement, script_pubkey) = match msg { ) -> Result<()> {
Confirmed { use collab_settlement_maker::Completed::*;
order_id, let (order_id, settlement, script_pubkey) = match msg {
settlement, Confirmed {
script_pubkey, order_id,
} => (order_id, settlement, script_pubkey), settlement,
Rejected { .. } => { script_pubkey,
return Ok(()); } => (order_id, settlement, script_pubkey),
} Rejected { .. } => {
Failed { order_id, error } => { return Ok(());
tracing::warn!(%order_id, "Collaborative settlement failed: {:#}", error); }
return Ok(()); Failed { order_id, error } => {
} tracing::warn!(%order_id, "Collaborative settlement failed: {:#}", error);
}; return Ok(());
}
};
let mut conn = self.db.acquire().await?; let mut conn = self.db.acquire().await?;
let mut cfd = load_cfd(order_id, &mut conn).await?; let mut cfd = load_cfd(order_id, &mut conn).await?;
let tx = settlement.tx.clone(); let tx = settlement.tx.clone();
cfd.handle_proposal_signed(settlement) cfd.handle_proposal_signed(settlement)
.context("Failed to update state with collaborative settlement")?; .context("Failed to update state with collaborative settlement")?;
append_cfd_state(&cfd, &mut conn, &self.projection_actor).await?; append_cfd_state(&cfd, &mut conn, &self.projection_actor).await?;
let txid = self let txid = self
.wallet .wallet
.send(wallet::TryBroadcastTransaction { tx }) .send(wallet::TryBroadcastTransaction { tx })
.await? .await?
.context("Broadcasting close transaction")?; .context("Broadcasting close transaction")?;
tracing::info!(%order_id, "Close transaction published with txid {}", txid); tracing::info!(%order_id, "Close transaction published with txid {}", txid);
self.monitor_actor self.monitor_actor
.send(monitor::CollaborativeSettlement { .send(monitor::CollaborativeSettlement {
order_id, order_id,
tx: (txid, script_pubkey), tx: (txid, script_pubkey),
}) })
.await?; .await?;
anyhow::Ok(()) Ok(())
});
} }
} }
@ -754,56 +754,54 @@ where
M: xtra::Handler<monitor::StartMonitoring>, M: xtra::Handler<monitor::StartMonitoring>,
W: xtra::Handler<wallet::TryBroadcastTransaction>, W: xtra::Handler<wallet::TryBroadcastTransaction>,
{ {
async fn handle_setup_completed(&mut self, msg: setup_maker::Completed) { async fn handle_setup_completed(&mut self, msg: setup_maker::Completed) -> Result<()> {
log_error!(async { use setup_maker::Completed::*;
use setup_maker::Completed::*; let (order_id, dlc) = match msg {
let (order_id, dlc) = match msg { NewContract { order_id, dlc } => (order_id, dlc),
NewContract { order_id, dlc } => (order_id, dlc), Failed { order_id, error } => {
Failed { order_id, error } => { self.append_cfd_state_setup_failed(order_id, error).await?;
self.append_cfd_state_setup_failed(order_id, error).await?; return anyhow::Ok(());
return anyhow::Ok(()); }
} Rejected(order_id) => {
Rejected(order_id) => { self.append_cfd_state_rejected(order_id).await?;
self.append_cfd_state_rejected(order_id).await?; return anyhow::Ok(());
return anyhow::Ok(()); }
} };
};
let mut conn = self.db.acquire().await?; let mut conn = self.db.acquire().await?;
let mut cfd = load_cfd(order_id, &mut conn).await?; let mut cfd = load_cfd(order_id, &mut conn).await?;
*cfd.state_mut() = CfdState::PendingOpen { *cfd.state_mut() = CfdState::PendingOpen {
common: CfdStateCommon::default(), common: CfdStateCommon::default(),
dlc: dlc.clone(), dlc: dlc.clone(),
attestation: None, attestation: None,
}; };
append_cfd_state(&cfd, &mut conn, &self.projection_actor).await?; append_cfd_state(&cfd, &mut conn, &self.projection_actor).await?;
let txid = self let txid = self
.wallet .wallet
.send(wallet::TryBroadcastTransaction { .send(wallet::TryBroadcastTransaction {
tx: dlc.lock.0.clone(), tx: dlc.lock.0.clone(),
}) })
.await??; .await??;
tracing::info!("Lock transaction published with txid {}", txid); tracing::info!("Lock transaction published with txid {}", txid);
self.monitor_actor self.monitor_actor
.send(monitor::StartMonitoring { .send(monitor::StartMonitoring {
id: order_id, id: order_id,
params: MonitorParams::new(dlc.clone(), cfd.refund_timelock_in_blocks()), params: MonitorParams::new(dlc.clone(), cfd.refund_timelock_in_blocks()),
}) })
.await?; .await?;
self.oracle_actor self.oracle_actor
.send(oracle::MonitorAttestation { .send(oracle::MonitorAttestation {
event_id: dlc.settlement_event_id, event_id: dlc.settlement_event_id,
}) })
.await?; .await?;
Ok(()) Ok(())
});
} }
} }
@ -812,8 +810,8 @@ impl<O: 'static, M: 'static, T: 'static, W: 'static> Handler<TakerConnected> for
where where
T: xtra::Handler<maker_inc_connections::TakerMessage>, T: xtra::Handler<maker_inc_connections::TakerMessage>,
{ {
async fn handle(&mut self, msg: TakerConnected, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: TakerConnected, _ctx: &mut Context<Self>) -> Result<()> {
log_error!(self.handle_taker_connected(msg.id)); self.handle_taker_connected(msg.id).await
} }
} }
@ -823,8 +821,8 @@ impl<O: 'static, M: 'static, T: 'static, W: 'static> Handler<TakerDisconnected>
where where
T: xtra::Handler<maker_inc_connections::TakerMessage>, T: xtra::Handler<maker_inc_connections::TakerMessage>,
{ {
async fn handle(&mut self, msg: TakerDisconnected, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: TakerDisconnected, _ctx: &mut Context<Self>) -> Result<()> {
log_error!(self.handle_taker_disconnected(msg.id)); self.handle_taker_disconnected(msg.id).await
} }
} }
@ -834,8 +832,8 @@ where
M: xtra::Handler<monitor::StartMonitoring>, M: xtra::Handler<monitor::StartMonitoring>,
O: xtra::Handler<oracle::MonitorAttestation>, O: xtra::Handler<oracle::MonitorAttestation>,
{ {
async fn handle(&mut self, msg: Completed, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: Completed, _ctx: &mut Context<Self>) -> Result<()> {
log_error!(self.handle_roll_over_completed(msg)); self.handle_roll_over_completed(msg).await
} }
} }
@ -844,8 +842,8 @@ impl<O: 'static, M: 'static, T: 'static, W: 'static> Handler<monitor::Event> for
where where
W: xtra::Handler<wallet::TryBroadcastTransaction>, W: xtra::Handler<wallet::TryBroadcastTransaction>,
{ {
async fn handle(&mut self, msg: monitor::Event, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: monitor::Event, _ctx: &mut Context<Self>) -> Result<()> {
log_error!(self.handle_monitoring_event(msg)) self.handle_monitoring_event(msg).await
} }
} }
@ -866,10 +864,15 @@ where
+ xtra::Handler<wallet::BuildPartyParams> + xtra::Handler<wallet::BuildPartyParams>
+ xtra::Handler<wallet::TryBroadcastTransaction>, + xtra::Handler<wallet::TryBroadcastTransaction>,
{ {
async fn handle(&mut self, FromTaker { taker_id, msg }: FromTaker, ctx: &mut Context<Self>) { async fn handle(
&mut self,
FromTaker { taker_id, msg }: FromTaker,
ctx: &mut Context<Self>,
) -> Result<()> {
match msg { match msg {
wire::TakerToMaker::TakeOrder { order_id, quantity } => { wire::TakerToMaker::TakeOrder { order_id, quantity } => {
log_error!(self.handle_take_order(taker_id, order_id, quantity, ctx)) self.handle_take_order(taker_id, order_id, quantity, ctx)
.await?
} }
wire::TakerToMaker::Settlement { wire::TakerToMaker::Settlement {
order_id, order_id,
@ -881,17 +884,22 @@ where
price, price,
}, },
} => { } => {
log_error!(self.handle_propose_settlement( if let Err(e) = self
taker_id, .handle_propose_settlement(
SettlementProposal { taker_id,
order_id, SettlementProposal {
timestamp, order_id,
taker, timestamp,
maker, taker,
price maker,
}, price,
ctx },
)) ctx,
)
.await
{
tracing::warn!("Failed ot handle settlement proposal: {:#}", e);
}
} }
wire::TakerToMaker::Settlement { wire::TakerToMaker::Settlement {
msg: wire::taker_to_maker::Settlement::Initiate { .. }, msg: wire::taker_to_maker::Settlement::Initiate { .. },
@ -903,14 +911,19 @@ where
order_id, order_id,
timestamp, timestamp,
} => { } => {
log_error!(self.handle_propose_roll_over( if let Err(e) = self
RolloverProposal { .handle_propose_roll_over(
order_id, RolloverProposal {
timestamp, order_id,
}, timestamp,
taker_id, },
ctx taker_id,
)) ctx,
)
.await
{
tracing::warn!("Failed to handle rollover proposal: {:#}", e);
}
} }
wire::TakerToMaker::RollOverProtocol { .. } => { wire::TakerToMaker::RollOverProtocol { .. } => {
unreachable!("This kind of message should be sent to the rollover_maker::Actor`") unreachable!("This kind of message should be sent to the rollover_maker::Actor`")
@ -921,7 +934,9 @@ where
TakerToMaker::Hello(_) => { TakerToMaker::Hello(_) => {
unreachable!("The Hello message is not sent to the cfd actor") unreachable!("The Hello message is not sent to the cfd actor")
} }
} };
Ok(())
} }
} }
@ -932,24 +947,26 @@ where
W: xtra::Handler<wallet::TryBroadcastTransaction>, W: xtra::Handler<wallet::TryBroadcastTransaction>,
{ {
async fn handle(&mut self, msg: oracle::Attestation, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: oracle::Attestation, _ctx: &mut Context<Self>) {
log_error!(self.handle_oracle_attestation(msg)) if let Err(e) = self.handle_oracle_attestation(msg).await {
tracing::warn!("Failed to handle oracle attestation: {:#}", e)
}
} }
} }
impl Message for TakerConnected { impl Message for TakerConnected {
type Result = (); type Result = Result<()>;
} }
impl Message for TakerDisconnected { impl Message for TakerDisconnected {
type Result = (); type Result = Result<()>;
} }
impl Message for Completed { impl Message for Completed {
type Result = (); type Result = Result<()>;
} }
impl Message for FromTaker { impl Message for FromTaker {
type Result = (); type Result = Result<()>;
} }
impl<O: 'static, M: 'static, T: 'static, W: 'static> xtra::Actor for Actor<O, M, T, W> {} impl<O: 'static, M: 'static, T: 'static, W: 'static> xtra::Actor for Actor<O, M, T, W> {}

9
daemon/src/maker_inc_connections.rs
View File

@ -20,6 +20,7 @@ use crate::wire::EncryptedJsonCodec;
use crate::wire::MakerToTaker; use crate::wire::MakerToTaker;
use crate::wire::TakerToMaker; use crate::wire::TakerToMaker;
use crate::wire::Version; use crate::wire::Version;
use crate::xtra_ext::LogFailure;
use crate::Tasks; use crate::Tasks;
use anyhow::bail; use anyhow::bail;
use anyhow::Context; use anyhow::Context;
@ -138,6 +139,7 @@ impl Actor {
let _ = self let _ = self
.taker_disconnected_channel .taker_disconnected_channel
.send(maker_cfd::TakerDisconnected { id: *taker_id }) .send(maker_cfd::TakerDisconnected { id: *taker_id })
.log_failure("Failed to inform about taker disconnect")
.await; .await;
let _ = self.connection_tasks.remove(taker_id); let _ = self.connection_tasks.remove(taker_id);
} }
@ -219,7 +221,7 @@ impl Actor {
let this = ctx.address().expect("self to be alive"); let this = ctx.address().expect("self to be alive");
let read_fut = async move { let read_fut = async move {
while let Ok(Some(msg)) = read.try_next().await { while let Ok(Some(msg)) = read.try_next().await {
let res = this.send(FromTaker { taker_id, msg }).await; let res = this.send(FromTaker { taker_id, msg }).log_failure("").await;
if res.is_err() { if res.is_err() {
break; break;
@ -249,6 +251,7 @@ impl Actor {
let _ = self let _ = self
.taker_connected_channel .taker_connected_channel
.send(maker_cfd::TakerConnected { id: taker_id }) .send(maker_cfd::TakerConnected { id: taker_id })
.log_failure("Failed to report new taker connection")
.await; .await;
Ok(()) Ok(())
@ -342,7 +345,7 @@ impl Actor {
#[xtra_productivity(message_impl = false)] #[xtra_productivity(message_impl = false)]
impl Actor { impl Actor {
async fn handle_msg_from_taker(&mut self, msg: FromTaker) { async fn handle_msg_from_taker(&mut self, msg: FromTaker) -> Result<()> {
use wire::TakerToMaker::*; use wire::TakerToMaker::*;
match msg.msg { match msg.msg {
Protocol { order_id, msg } => match self.setup_actors.get_connected(&order_id) { Protocol { order_id, msg } => match self.setup_actors.get_connected(&order_id) {
@ -380,6 +383,8 @@ impl Actor {
let _ = self.taker_msg_channel.send(msg); let _ = self.taker_msg_channel.send(msg);
} }
} }
Ok(())
} }
async fn handle_setup_actor_stopping(&mut self, message: Stopping<setup_maker::Actor>) { async fn handle_setup_actor_stopping(&mut self, message: Stopping<setup_maker::Actor>) {

19
daemon/src/monitor.rs
View File

@ -1,4 +1,3 @@
use crate::log_error;
use crate::model; use crate::model;
use crate::model::cfd::CetStatus; use crate::model::cfd::CetStatus;
use crate::model::cfd::Cfd; use crate::model::cfd::Cfd;
@ -333,7 +332,7 @@ where
Ok(()) Ok(())
} }
async fn handle_oracle_attestation(&mut self, attestation: oracle::Attestation) -> Result<()> { async fn handle_oracle_attestation(&mut self, attestation: oracle::Attestation) {
for (order_id, MonitorParams { cets, .. }) in self for (order_id, MonitorParams { cets, .. }) in self
.cfds .cfds
.clone() .clone()
@ -342,8 +341,6 @@ where
{ {
try_continue!(self.monitor_cet_finality(cets, attestation.clone(), order_id)) try_continue!(self.monitor_cet_finality(cets, attestation.clone(), order_id))
} }
Ok(())
} }
async fn update_state( async fn update_state(
@ -433,7 +430,7 @@ where
for (target_status, event) in reached_monitoring_target { for (target_status, event) in reached_monitoring_target {
tracing::info!(%txid, target = %target_status, current = %status, "Bitcoin transaction reached monitoring target"); tracing::info!(%txid, target = %target_status, current = %status, "Bitcoin transaction reached monitoring target");
self.event_channel.send(event).await?; self.event_channel.send(event).await??;
} }
} }
} }
@ -628,7 +625,7 @@ fn map_cets(
} }
impl xtra::Message for Event { impl xtra::Message for Event {
type Result = (); type Result = Result<()>;
} }
impl xtra::Message for Sync { impl xtra::Message for Sync {
@ -669,14 +666,16 @@ where
C: bdk::electrum_client::ElectrumApi + Send + 'static, C: bdk::electrum_client::ElectrumApi + Send + 'static,
{ {
async fn handle(&mut self, _: Sync, _ctx: &mut xtra::Context<Self>) { async fn handle(&mut self, _: Sync, _ctx: &mut xtra::Context<Self>) {
log_error!(self.sync()); if let Err(e) = self.sync().await {
tracing::warn!("Sync failed: {:#}", e);
}
} }
} }
#[async_trait] #[async_trait]
impl xtra::Handler<oracle::Attestation> for Actor { impl xtra::Handler<oracle::Attestation> for Actor {
async fn handle(&mut self, msg: oracle::Attestation, _ctx: &mut xtra::Context<Self>) { async fn handle(&mut self, msg: oracle::Attestation, _ctx: &mut xtra::Context<Self>) {
log_error!(self.handle_oracle_attestation(msg)); self.handle_oracle_attestation(msg).await
} }
} }
@ -852,8 +851,10 @@ mod tests {
#[async_trait] #[async_trait]
impl xtra::Handler<Event> for MessageRecordingActor { impl xtra::Handler<Event> for MessageRecordingActor {
async fn handle(&mut self, message: Event, _ctx: &mut xtra::Context<Self>) { async fn handle(&mut self, message: Event, _ctx: &mut xtra::Context<Self>) -> Result<()> {
self.events.push(message); self.events.push(message);
Ok(())
} }
} }

14
daemon/src/oracle.rs
View File

@ -1,9 +1,9 @@
use crate::log_error;
use crate::model::cfd::Cfd; use crate::model::cfd::Cfd;
use crate::model::cfd::CfdState; use crate::model::cfd::CfdState;
use crate::model::BitMexPriceEventId; use crate::model::BitMexPriceEventId;
use crate::tokio_ext; use crate::tokio_ext;
use crate::try_continue; use crate::try_continue;
use crate::xtra_ext::LogFailure;
use anyhow::Context; use anyhow::Context;
use anyhow::Result; use anyhow::Result;
use async_trait::async_trait; use async_trait::async_trait;
@ -188,6 +188,7 @@ impl Actor {
id: event_id, id: event_id,
attestation, attestation,
}) })
.log_failure("Failed to send attestation to oracle::Actor")
.await?; .await?;
Ok(()) Ok(())
@ -257,8 +258,13 @@ pub struct NoAnnouncement(pub BitMexPriceEventId);
#[async_trait] #[async_trait]
impl xtra::Handler<NewAttestationFetched> for Actor { impl xtra::Handler<NewAttestationFetched> for Actor {
async fn handle(&mut self, msg: NewAttestationFetched, _ctx: &mut xtra::Context<Self>) { async fn handle(
log_error!(self.handle_new_attestation_fetched(msg.id, msg.attestation)); &mut self,
msg: NewAttestationFetched,
_ctx: &mut xtra::Context<Self>,
) -> Result<()> {
self.handle_new_attestation_fetched(msg.id, msg.attestation)
.await
} }
} }
@ -305,7 +311,7 @@ impl xtra::Message for Attestation {
} }
impl xtra::Message for NewAttestationFetched { impl xtra::Message for NewAttestationFetched {
type Result = (); type Result = Result<()>;
} }
mod olivia_api { mod olivia_api {

6
daemon/src/rollover_maker.rs
View File

@ -20,6 +20,7 @@ use crate::tokio_ext::spawn_fallible;
use crate::wire; use crate::wire;
use crate::wire::MakerToTaker; use crate::wire::MakerToTaker;
use crate::wire::RollOverMsg; use crate::wire::RollOverMsg;
use crate::xtra_ext::LogFailure;
use crate::Cfd; use crate::Cfd;
use crate::Stopping; use crate::Stopping;
use anyhow::Context as _; use anyhow::Context as _;
@ -140,7 +141,10 @@ impl Actor {
order_id: self.cfd.id(), order_id: self.cfd.id(),
dlc, dlc,
}; };
self.maker_cfd_actor.send(msg).await?; self.maker_cfd_actor
.send(msg)
.log_failure("Failed to report rollover completion")
.await?;
ctx.stop(); ctx.stop();
Ok(()) Ok(())
} }

7
daemon/src/setup_maker.rs
View File

@ -14,6 +14,7 @@ use crate::tokio_ext::spawn_fallible;
use crate::wallet; use crate::wallet;
use crate::wire; use crate::wire;
use crate::wire::SetupMsg; use crate::wire::SetupMsg;
use crate::xtra_ext::LogFailure;
use anyhow::Context; use anyhow::Context;
use anyhow::Result; use anyhow::Result;
use async_trait::async_trait; use async_trait::async_trait;
@ -121,7 +122,11 @@ impl Actor {
} }
async fn complete(&mut self, completed: Completed, ctx: &mut xtra::Context<Self>) { async fn complete(&mut self, completed: Completed, ctx: &mut xtra::Context<Self>) {
let _ = self.on_completed.send(completed).await; let _ = self
.on_completed
.send(completed)
.log_failure("Failed to inform about contract setup completion")
.await;
ctx.stop(); ctx.stop();
} }

13
daemon/src/setup_taker.rs
View File

@ -13,6 +13,7 @@ use crate::tokio_ext::spawn_fallible;
use crate::wallet; use crate::wallet;
use crate::wire; use crate::wire;
use crate::wire::SetupMsg; use crate::wire::SetupMsg;
use crate::xtra_ext::LogFailure;
use anyhow::Context; use anyhow::Context;
use anyhow::Result; use anyhow::Result;
use async_trait::async_trait; use async_trait::async_trait;
@ -74,7 +75,10 @@ impl Actor {
// inform the `taker_cfd::Actor` about the start of contract // inform the `taker_cfd::Actor` about the start of contract
// setup, so that the db and UI can be updated accordingly // setup, so that the db and UI can be updated accordingly
self.on_accepted.send(Started(order_id)).await?; self.on_accepted
.send(Started(order_id))
.log_failure("Failed to inform about contract setup start")
.await?;
let (sender, receiver) = mpsc::unbounded::<SetupMsg>(); let (sender, receiver) = mpsc::unbounded::<SetupMsg>();
// store the writing end to forward messages from the maker to // store the writing end to forward messages from the maker to
@ -124,6 +128,7 @@ impl Actor {
self.on_completed self.on_completed
.send(Completed::Rejected { order_id }) .send(Completed::Rejected { order_id })
.log_failure("Failed to inform about contract setup rejection")
.await?; .await?;
ctx.stop(); ctx.stop();
@ -147,6 +152,7 @@ impl Actor {
order_id: msg.order_id, order_id: msg.order_id,
dlc: msg.dlc, dlc: msg.dlc,
}) })
.log_failure("Failed to inform about contract setup completion")
.await?; .await?;
ctx.stop(); ctx.stop();
@ -160,6 +166,7 @@ impl Actor {
order_id: msg.order_id, order_id: msg.order_id,
error: msg.error, error: msg.error,
}) })
.log_failure("Failed to inform about contract setup failure")
.await?; .await?;
ctx.stop(); ctx.stop();
@ -240,11 +247,11 @@ impl Rejected {
} }
impl xtra::Message for Started { impl xtra::Message for Started {
type Result = (); type Result = Result<()>;
} }
impl xtra::Message for Completed { impl xtra::Message for Completed {
type Result = (); type Result = Result<()>;
} }
impl address_map::ActorName for Actor { impl address_map::ActorName for Actor {

21
daemon/src/taker_cfd.rs
View File

@ -5,7 +5,6 @@ use crate::cfd_actors::{self};
use crate::collab_settlement_taker; use crate::collab_settlement_taker;
use crate::connection; use crate::connection;
use crate::db::load_cfd; use crate::db::load_cfd;
use crate::log_error;
use crate::model::cfd::Cfd; use crate::model::cfd::Cfd;
use crate::model::cfd::CfdState; use crate::model::cfd::CfdState;
use crate::model::cfd::CfdStateCommon; use crate::model::cfd::CfdStateCommon;
@ -412,8 +411,8 @@ where
#[xtra_productivity] #[xtra_productivity]
impl<O, M, W> Actor<O, M, W> { impl<O, M, W> Actor<O, M, W> {
async fn handle_current_order(&mut self, msg: CurrentOrder) { async fn handle_current_order(&mut self, msg: CurrentOrder) -> Result<()> {
log_error!(self.handle_new_order(msg.0)); self.handle_new_order(msg.0).await
} }
} }
@ -424,8 +423,8 @@ where
M: xtra::Handler<monitor::StartMonitoring>, M: xtra::Handler<monitor::StartMonitoring>,
W: xtra::Handler<wallet::TryBroadcastTransaction>, W: xtra::Handler<wallet::TryBroadcastTransaction>,
{ {
async fn handle(&mut self, msg: Completed, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: Completed, _ctx: &mut Context<Self>) -> Result<()> {
log_error!(self.handle_setup_completed(msg)) self.handle_setup_completed(msg).await
} }
} }
@ -434,8 +433,8 @@ impl<O: 'static, M: 'static, W: 'static> Handler<monitor::Event> for Actor<O, M,
where where
W: xtra::Handler<wallet::TryBroadcastTransaction>, W: xtra::Handler<wallet::TryBroadcastTransaction>,
{ {
async fn handle(&mut self, msg: monitor::Event, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: monitor::Event, _ctx: &mut Context<Self>) -> Result<()> {
log_error!(self.handle_monitoring_event(msg)) self.handle_monitoring_event(msg).await
} }
} }
@ -445,14 +444,16 @@ where
W: xtra::Handler<wallet::TryBroadcastTransaction>, W: xtra::Handler<wallet::TryBroadcastTransaction>,
{ {
async fn handle(&mut self, msg: oracle::Attestation, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: oracle::Attestation, _ctx: &mut Context<Self>) {
log_error!(self.handle_oracle_attestation(msg)) if let Err(e) = self.handle_oracle_attestation(msg).await {
tracing::warn!("Failed to handle oracle attestation: {:#}", e)
}
} }
} }
#[async_trait] #[async_trait]
impl<O: 'static, M: 'static, W: 'static> Handler<setup_taker::Started> for Actor<O, M, W> { impl<O: 'static, M: 'static, W: 'static> Handler<setup_taker::Started> for Actor<O, M, W> {
async fn handle(&mut self, msg: setup_taker::Started, _ctx: &mut Context<Self>) { async fn handle(&mut self, msg: setup_taker::Started, _ctx: &mut Context<Self>) -> Result<()> {
log_error!(self.handle_setup_started(msg.0)) self.handle_setup_started(msg.0).await
} }
} }

50
daemon/src/xtra_ext.rs
View File

@ -0,0 +1,50 @@
use async_trait::async_trait;
use xtra::address;
use xtra::message_channel;
use xtra::Actor;
use xtra::Disconnected;
use xtra::Message;
#[async_trait]
pub trait LogFailure {
async fn log_failure(self, context: &str) -> Result<(), Disconnected>;
}
#[async_trait]
impl<A, M> LogFailure for address::SendFuture<A, M>
where
A: Actor,
M: Message<Result = anyhow::Result<()>>,
{
async fn log_failure(self, context: &str) -> Result<(), Disconnected> {
if let Err(e) = self.await? {
tracing::warn!(
"{}: Message handler for message {} failed: {:#}",
context,
std::any::type_name::<M>(),
e
);
}
Ok(())
}
}
#[async_trait]
impl<M> LogFailure for message_channel::SendFuture<M>
where
M: xtra::Message<Result = anyhow::Result<()>>,
{
async fn log_failure(self, context: &str) -> Result<(), Disconnected> {
if let Err(e) = self.await? {
tracing::warn!(
"{}: Message handler for message {} failed: {:#}",
context,
std::any::type_name::<M>(),
e
);
}
Ok(())
}
}

5
daemon/tests/harness/mocks/monitor.rs
View File

@ -1,8 +1,7 @@
use std::sync::Arc;
use daemon::monitor; use daemon::monitor;
use daemon::oracle; use daemon::oracle;
use mockall::*; use mockall::*;
use std::sync::Arc;
use tokio::sync::Mutex; use tokio::sync::Mutex;
use xtra_productivity::xtra_productivity; use xtra_productivity::xtra_productivity;
@ -30,7 +29,7 @@ impl MonitorActor {
} }
async fn handle(&mut self, msg: oracle::Attestation) { async fn handle(&mut self, msg: oracle::Attestation) {
self.mock.lock().await.oracle_attestation(msg) self.mock.lock().await.oracle_attestation(msg);
} }
} }

10
daemon/tests/harness/mod.rs
View File

@ -367,9 +367,13 @@ impl Taker {
#[macro_export] #[macro_export]
macro_rules! deliver_event { macro_rules! deliver_event {
($maker:expr, $taker:expr, $event:expr) => { ($maker:expr, $taker:expr, $event:expr) => {
tracing::debug!("Delivering event: {:?}", $event); #[allow(unused_must_use)]
$taker.system.cfd_actor_addr.send($event).await.unwrap(); {
$maker.system.cfd_actor_addr.send($event).await.unwrap(); tracing::debug!("Delivering event: {:?}", $event);
$taker.system.cfd_actor_addr.send($event).await.unwrap();
$maker.system.cfd_actor_addr.send($event).await.unwrap();
}
}; };
} }

Write Preview
Loading…
Cancel
Save