Browse Source
967: Introduce process manager actor r=da-kami a=da-kami Resolves #956 1. Event Saving: Only the process manager saves events into the dabase. The initial CFD creation (prior to events) is done in the cfd actors. 2. Event post-processing: Only the process manager handles post-processing of events 3. UI updates through projection: Only the process manager updates the UI after an event is generated. (Note: Upon initial cfd creation the UI is updated through the cfd actor when inserting the cfd; Upon startup the projection actor updates itself) As a result of this the cfd actor now does not know about the monitor anymore; and furthermore we don't have to keep bounds for wallet/oracle on most of the function calls that are handled by the process manager. This massively reduces the trait bounds and impl blocks in the cfd actors. Co-authored-by: Daniel Karzel <daniel@comit.network>master preview
bors[bot]
3 years ago
committed by
GitHub
GitHub
9 changed files with 419 additions and 499 deletions
@ -0,0 +1,177 @@ |
|||
use crate::db::append_event; |
|||
use crate::model::cfd; |
|||
use crate::model::cfd::CfdEvent; |
|||
use crate::model::cfd::Role; |
|||
use crate::monitor; |
|||
use crate::monitor::MonitorParams; |
|||
use crate::oracle; |
|||
use crate::projection; |
|||
use crate::wallet; |
|||
use anyhow::Context; |
|||
use anyhow::Result; |
|||
use xtra::prelude::MessageChannel; |
|||
use xtra_productivity::xtra_productivity; |
|||
|
|||
pub struct Actor { |
|||
db: sqlx::SqlitePool, |
|||
role: Role, |
|||
cfds_changed: Box<dyn MessageChannel<projection::CfdsChanged>>, |
|||
try_broadcast_transaction: Box<dyn MessageChannel<wallet::TryBroadcastTransaction>>, |
|||
start_monitoring: Box<dyn MessageChannel<monitor::StartMonitoring>>, |
|||
monitor_collaborative_settlement: Box<dyn MessageChannel<monitor::CollaborativeSettlement>>, |
|||
monitor_attestation: Box<dyn MessageChannel<oracle::MonitorAttestation>>, |
|||
} |
|||
|
|||
pub struct Event(cfd::Event); |
|||
|
|||
impl Event { |
|||
pub fn new(event: cfd::Event) -> Self { |
|||
Self(event) |
|||
} |
|||
} |
|||
|
|||
impl Actor { |
|||
pub fn new( |
|||
db: sqlx::SqlitePool, |
|||
role: Role, |
|||
cfds_changed: &(impl MessageChannel<projection::CfdsChanged> + 'static), |
|||
try_broadcast_transaction: &(impl MessageChannel<wallet::TryBroadcastTransaction> + 'static), |
|||
start_monitoring: &(impl MessageChannel<monitor::StartMonitoring> + 'static), |
|||
monitor_collaborative_settlement: &(impl MessageChannel<monitor::CollaborativeSettlement> |
|||
+ 'static), |
|||
monitor_attestation: &(impl MessageChannel<oracle::MonitorAttestation> + 'static), |
|||
) -> Self { |
|||
Self { |
|||
db, |
|||
role, |
|||
cfds_changed: cfds_changed.clone_channel(), |
|||
try_broadcast_transaction: try_broadcast_transaction.clone_channel(), |
|||
start_monitoring: start_monitoring.clone_channel(), |
|||
monitor_collaborative_settlement: monitor_collaborative_settlement.clone_channel(), |
|||
monitor_attestation: monitor_attestation.clone_channel(), |
|||
} |
|||
} |
|||
} |
|||
|
|||
#[xtra_productivity] |
|||
impl Actor { |
|||
fn handle(&mut self, msg: Event) -> Result<()> { |
|||
let event = msg.0; |
|||
|
|||
// 1. Safe in DB
|
|||
let mut conn = self.db.acquire().await?; |
|||
append_event(event.clone(), &mut conn).await?; |
|||
|
|||
// 2. Post process event
|
|||
match event.event { |
|||
CfdEvent::ContractSetupCompleted { dlc } => { |
|||
tracing::info!("Setup complete, publishing on chain now"); |
|||
|
|||
let lock_tx = dlc.lock.0.clone(); |
|||
let txid = self |
|||
.try_broadcast_transaction |
|||
.send(wallet::TryBroadcastTransaction { tx: lock_tx }) |
|||
.await??; |
|||
|
|||
tracing::info!("Lock transaction published with txid {}", txid); |
|||
|
|||
self.start_monitoring |
|||
.send(monitor::StartMonitoring { |
|||
id: event.id, |
|||
params: MonitorParams::new(dlc.clone()), |
|||
}) |
|||
.await?; |
|||
|
|||
self.monitor_attestation |
|||
.send(oracle::MonitorAttestation { |
|||
event_id: dlc.settlement_event_id, |
|||
}) |
|||
.await?; |
|||
} |
|||
CfdEvent::CollaborativeSettlementCompleted { |
|||
spend_tx, script, .. |
|||
} => { |
|||
let txid = match self.role { |
|||
Role::Maker => { |
|||
let txid = self |
|||
.try_broadcast_transaction |
|||
.send(wallet::TryBroadcastTransaction { tx: spend_tx }) |
|||
.await? |
|||
.context("Broadcasting close transaction")?; |
|||
|
|||
tracing::info!(order_id=%event.id, "Close transaction published with txid {}", txid); |
|||
|
|||
txid |
|||
} |
|||
Role::Taker => { |
|||
// TODO: Publish the tx once the collaborative settlement is symmetric,
|
|||
// allowing the taker to publish as well.
|
|||
let txid = spend_tx.txid(); |
|||
tracing::info!(order_id=%event.id, "Collaborative settlement completed successfully {}", txid); |
|||
txid |
|||
} |
|||
}; |
|||
|
|||
self.monitor_collaborative_settlement |
|||
.send(monitor::CollaborativeSettlement { |
|||
order_id: event.id, |
|||
tx: (txid, script), |
|||
}) |
|||
.await?; |
|||
} |
|||
CfdEvent::CollaborativeSettlementRejected { commit_tx } => { |
|||
let txid = self |
|||
.try_broadcast_transaction |
|||
.send(wallet::TryBroadcastTransaction { tx: commit_tx }) |
|||
.await? |
|||
.context("Broadcasting commit transaction")?; |
|||
|
|||
tracing::info!( |
|||
"Closing non-collaboratively. Commit tx published with txid {}", |
|||
txid |
|||
) |
|||
} |
|||
CfdEvent::CollaborativeSettlementFailed { commit_tx } => { |
|||
let txid = self |
|||
.try_broadcast_transaction |
|||
.send(wallet::TryBroadcastTransaction { tx: commit_tx }) |
|||
.await? |
|||
.context("Broadcasting commit transaction")?; |
|||
|
|||
tracing::warn!( |
|||
"Closing non-collaboratively. Commit tx published with txid {}", |
|||
txid |
|||
) |
|||
} |
|||
CfdEvent::OracleAttestedPostCetTimelock { cet, .. } |
|||
| CfdEvent::CetTimelockConfirmedPostOracleAttestation { cet } => { |
|||
let txid = self |
|||
.try_broadcast_transaction |
|||
.send(wallet::TryBroadcastTransaction { tx: cet }) |
|||
.await? |
|||
.context("Failed to broadcast CET")?; |
|||
|
|||
tracing::info!(%txid, "CET published"); |
|||
} |
|||
CfdEvent::OracleAttestedPriorCetTimelock { commit_tx: tx, .. } |
|||
| CfdEvent::ManualCommit { tx } => { |
|||
let txid = self |
|||
.try_broadcast_transaction |
|||
.send(wallet::TryBroadcastTransaction { tx }) |
|||
.await? |
|||
.context("Failed to broadcast commit transaction")?; |
|||
|
|||
tracing::info!(%txid, "Commit transaction published"); |
|||
} |
|||
|
|||
_ => {} // TODO: Monitor post processing for rollover
|
|||
} |
|||
|
|||
// 3. Update UI
|
|||
self.cfds_changed.send(projection::CfdsChanged).await?; |
|||
|
|||
Ok(()) |
|||
} |
|||
} |
|||
|
|||
impl xtra::Actor for Actor {} |
|||
Loading…
Reference in new issue