Monitor for close finality

After starting collaborative settlement,

daemon/src/maker_cfd.rs

@@ -5,8 +5,7 @@ use crate::db::{
 use crate::maker_inc_connections::TakerCommand;
 use crate::model::cfd::{
     Attestation, Cfd, CfdState, CfdStateChangeEvent, CfdStateCommon, Dlc, Order, OrderId, Origin,
-    Role, RollOverProposal, SettlementKind, SettlementProposal, UpdateCfdProposal,
-    UpdateCfdProposals,
+   Role, RollOverProposal, SettlementKind, SettlementProposal, UpdateCfdProposal, UpdateCfdProposals,
 };
 use crate::model::{TakerId, Usd};
 use crate::monitor::MonitorParams;
@@ -298,23 +297,34 @@ impl Actor {
 
         let mut conn = self.db.acquire().await?;
 
-        let cfd = load_cfd_by_order_id(order_id, &mut conn).await?;
+        let mut cfd = load_cfd_by_order_id(order_id, &mut conn).await?;
         let dlc = cfd.open_dlc().context("CFD was in wrong state")?;
 
         let (tx, sig_maker) = dlc.close_transaction(proposal)?;
+
+        cfd.handle(CfdStateChangeEvent::ProposalSigned(
+            TimestampedTransaction::new(tx.clone()),
+        ))?;
+        insert_new_cfd_state_by_order_id(cfd.order.id, cfd.state.clone(), &mut conn).await?;
+
         let spend_tx = dlc.finalize_spend_transaction((tx, sig_maker), sig_taker)?;
 
-        self.wallet
-            .try_broadcast_transaction(spend_tx)
+        let txid = self
+            .wallet
+            .try_broadcast_transaction(spend_tx.clone())
             .await
             .context("Broadcasting spend transaction")?;
+        tracing::info!("Close transaction published with txid {}", txid);
+
+        cfd.handle(CfdStateChangeEvent::CloseSent(TimestampedTransaction::new(
+            spend_tx,
+        )))?;
+        insert_new_cfd_state_by_order_id(cfd.order.id, cfd.state, &mut conn).await?;
 
         self.current_agreed_proposals
             .remove(&order_id)
             .context("remove accepted proposal after signing")?;
 
-        // TODO: Monitor for the transaction
-
         Ok(())
     }
 
@@ -461,6 +471,7 @@ impl Actor {
                 },
                 dlc: dlc.clone(),
                 attestation: None,
+                collaborative_close: None,
             },
             &mut conn,
         )

daemon/src/model/cfd.rs

@@ -216,6 +216,7 @@ pub enum CfdState {
         common: CfdStateCommon,
         dlc: Dlc,
         attestation: Option<Attestation>,
+        collaborative_close: Option<TimestampedTransaction>,
     },
 
     /// The commit transaction was published but it not final yet
@@ -251,6 +252,17 @@ pub enum CfdState {
         attestation: Attestation,
     },
 
+    /// The collaborative close transaction was published but is not final yet.
+    ///
+    /// This state applies to taker and maker.
+    /// This state is needed, because otherwise the user does not get any feedback.
+    PendingClose {
+        common: CfdStateCommon,
+        dlc: Dlc,
+        attestation: Option<Attestation>,
+        collaborative_close: TimestampedTransaction,
+    },
+
     /// The position was closed collaboratively or non-collaboratively
     ///
     /// This state applies to taker and maker.
@@ -260,7 +272,8 @@ pub enum CfdState {
     /// commit + cet).
     Closed {
         common: CfdStateCommon,
-        attestation: Attestation,
+        // TODO: Use an enum of either Attestation or CollaborativeSettlement
+        attestation: Option<Attestation>,
     },
 
     // TODO: Can be extended with CetStatus
@@ -391,6 +404,7 @@ impl CfdState {
             CfdState::SetupFailed { common, .. } => common,
             CfdState::PendingCommit { common, .. } => common,
             CfdState::PendingCet { common, .. } => common,
+            CfdState::PendingClose { common, .. } => common,
             CfdState::Closed { common, .. } => common,
         };
 
@@ -400,6 +414,20 @@ impl CfdState {
     pub fn get_transition_timestamp(&self) -> SystemTime {
         self.get_common().transition_timestamp
     }
+
+    pub fn get_collaborative_close(&self) -> Option<TimestampedTransaction> {
+        match self {
+            CfdState::Open {
+                collaborative_close,
+                ..
+            } => collaborative_close.clone(),
+            CfdState::PendingClose {
+                collaborative_close,
+                ..
+            } => Some(collaborative_close.clone()),
+            _ => None,
+        }
+    }
 }
 
 impl fmt::Display for CfdState {
@@ -444,6 +472,9 @@ impl fmt::Display for CfdState {
             CfdState::PendingCet { .. } => {
                 write!(f, "Pending CET")
             }
+            CfdState::PendingClose { .. } => {
+                write!(f, "Pending Close")
+            }
             CfdState::Closed { .. } => {
                 write!(f, "Closed")
             }
@@ -531,7 +562,7 @@ impl Cfd {
 
     pub fn profit(&self, current_price: Usd) -> Result<(SignedAmount, Percent)> {
         // TODO: We should use the payout curve here and not just the current price!
-
+        // TODO: Use the collab settlement if there was one
         let current_price = if let Some(attestation) = self.attestation() {
             attestation.price()
         } else {
@@ -626,9 +657,13 @@ impl Cfd {
                             },
                             dlc,
                             attestation: None,
+                            collaborative_close: None,
                         }
                     } else if let Open {
-                        dlc, attestation, ..
+                        dlc,
+                        attestation,
+                        collaborative_close,
+                        ..
                     } = self.state.clone()
                     {
                         CfdState::Open {
@@ -637,6 +672,7 @@ impl Cfd {
                             },
                             dlc,
                             attestation,
+                            collaborative_close,
                         }
                     } else {
                         bail!(
@@ -679,6 +715,12 @@ impl Cfd {
                         },
                     }
                 }
+                monitor::Event::CloseFinality(_) => CfdState::Closed {
+                    common: CfdStateCommon {
+                        transition_timestamp: SystemTime::now(),
+                    },
+                    attestation: None
+                },
                 monitor::Event::CetTimelockExpired(_) => match self.state.clone() {
                     CfdState::OpenCommitted {
                         dlc,
@@ -769,7 +811,7 @@ impl Cfd {
                         common: CfdStateCommon {
                             transition_timestamp: SystemTime::now(),
                         },
-                        attestation,
+                        attestation: Some(attestation),
                     }
                 }
                 monitor::Event::RevokedTransactionFound(_) => {
@@ -777,19 +819,21 @@ impl Cfd {
                 }
             },
             CfdStateChangeEvent::CommitTxSent => {
-                let (dlc, attestation) = if let PendingOpen {
-                    dlc, attestation, ..
-                }
-                | Open {
-                    dlc, attestation, ..
-                } = self.state.clone()
-                {
-                    (dlc, attestation)
-                } else {
-                    bail!(
-                        "Cannot transition to PendingCommit because of unexpected state {}",
-                        self.state
-                    )
+                let (dlc, attestation ) = match self.state.clone() {
+                    PendingOpen {
+                        dlc, attestation, ..
+                    } => (dlc, attestation),
+                    Open {
+                        dlc,
+                        attestation,
+                        ..
+                    } => (dlc, attestation),
+                    _ => {
+                        bail!(
+                            "Cannot transition to PendingCommit because of unexpected state {}",
+                            self.state
+                        )
+                    }
                 };
 
                 PendingCommit {
@@ -814,8 +858,12 @@ impl Cfd {
                     },
                     dlc,
                     attestation: Some(attestation),
+                    collaborative_close: None,
                 },
-                CfdState::PendingCommit { dlc, .. } => CfdState::PendingCommit {
+                CfdState::PendingCommit {
+                    dlc,
+                    ..
+                } => CfdState::PendingCommit {
                     common: CfdStateCommon {
                         transition_timestamp: SystemTime::now(),
                     },
@@ -862,6 +910,43 @@ impl Cfd {
                     dlc,
                     attestation,
                 }
+               
+            },
+            CfdStateChangeEvent::ProposalSigned(collaborative_close) => match self.state.clone() {
+                CfdState::Open {
+                    common,
+                    dlc,
+                    attestation,
+                    ..
+                } => CfdState::Open {
+                    common,
+                    dlc,
+                    attestation,
+                    collaborative_close: Some(collaborative_close),
+                },
+                _ => bail!(
+                    "Cannot add proposed settlement details to state because of unexpected state {}",
+                    self.state
+                ),
+            },
+            CfdStateChangeEvent::CloseSent(collaborative_close) => match self.state.clone() {
+                CfdState::Open {
+                    common,
+                    dlc,
+                    attestation,
+                    collaborative_close : Some(_),
+                } => CfdState::PendingClose {
+                    common,
+                    dlc,
+                    attestation,
+                    collaborative_close,
+                },
+                CfdState::Open {
+                    collaborative_close : None,
+                    ..
+                } => bail!("Cannot transition to PendingClose because Open state did not record a settlement proposal beforehand"),
+                _ => bail!(
+                    "Cannot transition to PendingClose because of unexpected state {}",
             }
         };
 
@@ -1058,6 +1143,7 @@ impl Cfd {
             | CfdState::Accepted { .. }
             | CfdState::Rejected { .. }
             | CfdState::ContractSetup { .. }
+            | CfdState::PendingClose { .. }
             | CfdState::Closed { .. }
             | CfdState::MustRefund { .. }
             | CfdState::Refunded { .. }
@@ -1084,7 +1170,10 @@ impl Cfd {
                 ..
             }
             | CfdState::PendingCet { attestation, .. }
-            | CfdState::Closed { attestation, .. } => Some(attestation),
+            | CfdState::Closed {
+                attestation: Some(attestation),
+                ..
+            } => Some(attestation),
 
             CfdState::OutgoingOrderRequest { .. }
             | CfdState::IncomingOrderRequest { .. }
@@ -1094,6 +1183,8 @@ impl Cfd {
             | CfdState::PendingOpen { .. }
             | CfdState::Open { .. }
             | CfdState::PendingCommit { .. }
+            | CfdState::PendingClose { .. }
+            | CfdState::Closed { .. }
             | CfdState::OpenCommitted { .. }
             | CfdState::MustRefund { .. }
             | CfdState::Refunded { .. }
@@ -1109,6 +1200,7 @@ pub struct NotReadyYet {
 }
 
 #[derive(Debug, Clone)]
+#[allow(dead_code)] // Not all variants are used by all binaries.
 pub enum CfdStateChangeEvent {
     // TODO: group other events by actors into enums and add them here so we can bundle all
     // transitions into cfd.transition_to(...)
@@ -1116,6 +1208,10 @@ pub enum CfdStateChangeEvent {
     CommitTxSent,
     OracleAttestation(Attestation),
     CetSent,
+    /// Settlement proposal was signed by the taker and sent to the maker
+    ProposalSigned(TimestampedTransaction),
+    /// Maker signed and finalized the close transaction with both signatures
+    CloseSent(TimestampedTransaction),
 }
 
 /// Returns the Profit/Loss (P/L) as Bitcoin. Losses are capped by the provided margin
@@ -1577,3 +1673,22 @@ pub struct RevokedCommit {
     pub txid: Txid,
     pub script_pubkey: Script,
 }
+
+/// Used when transactions (e.g. collaborative close) are recorded as a part of
+/// CfdState in the cases when we can't solely rely on state transition
+/// timestamp as it could have occured for different reasons (like a new
+/// attestation in Open state)
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+pub struct TimestampedTransaction {
+    pub tx: Transaction,
+    pub timestamp: SystemTime,
+}
+
+impl TimestampedTransaction {
+    pub fn new(tx: Transaction) -> Self {
+        Self {
+            tx,
+            timestamp: SystemTime::now(),
+        }
+    }
+}

daemon/src/monitor.rs

@@ -1,4 +1,4 @@
-use crate::model::cfd::{CetStatus, Cfd, CfdState, Dlc, OrderId};
+use crate::model::cfd::{CetStatus, Cfd, CfdState, Dlc, OrderId, TimestampedTransaction};
 use crate::model::BitMexPriceEventId;
 use crate::oracle::Attestation;
 use crate::{log_error, model, oracle};
@@ -89,6 +89,13 @@ where
                     actor.monitor_commit_cet_timelock(&params, cfd.order.id);
                     actor.monitor_commit_refund_timelock(&params, cfd.order.id);
                     actor.monitor_refund_finality(&params,cfd.order.id);
+
+                    if let Some(TimestampedTransaction { tx, ..}
+                    ) = cfd.state.get_collaborative_close()  {
+                        let close_params = (tx.txid(),
+                            tx.output.first().expect("have output").script_pubkey.clone());
+                        actor.monitor_close_finality(close_params,cfd.order.id);
+                    }
                 }
                 CfdState::OpenCommitted { dlc, cet_status, .. } => {
                     let params = MonitorParams::from_dlc_and_timelocks(dlc.clone(), cfd.refund_timelock_in_blocks());
@@ -124,6 +131,12 @@ where
                     actor.monitor_cet_finality(map_cets(dlc.cets), attestation.into(), cfd.order.id)?;
                     actor.monitor_commit_refund_timelock(&params, cfd.order.id);
                     actor.monitor_refund_finality(&params,cfd.order.id);
+	        }
+                CfdState::PendingClose { collaborative_close, .. } => {
+                    let transaction  = collaborative_close.tx;
+                    let close_params = (transaction.txid(),
+                                        transaction.output.first().expect("have output").script_pubkey.clone());
+                    actor.monitor_close_finality(close_params,cfd.order.id);
                 }
                 CfdState::MustRefund { dlc, .. } => {
                     let params = MonitorParams::from_dlc_and_timelocks(dlc.clone(), cfd.refund_timelock_in_blocks());
@@ -179,6 +192,13 @@ where
             .push((ScriptStatus::finality(), Event::CommitFinality(order_id)));
     }
 
+    fn monitor_close_finality(&mut self, close_params: (Txid, Script), order_id: OrderId) {
+        self.awaiting_status
+            .entry(close_params)
+            .or_default()
+            .push((ScriptStatus::finality(), Event::CloseFinality(order_id)));
+    }
+
     fn monitor_commit_cet_timelock(&mut self, params: &MonitorParams, order_id: OrderId) {
         self.awaiting_status
             .entry((params.commit.0, params.commit.1.script_pubkey()))
@@ -516,6 +536,7 @@ impl xtra::Message for StartMonitoring {
 pub enum Event {
     LockFinality(OrderId),
     CommitFinality(OrderId),
+    CloseFinality(OrderId),
     CetTimelockExpired(OrderId),
     CetFinality(OrderId),
     RefundTimelockExpired(OrderId),
@@ -528,6 +549,7 @@ impl Event {
         let order_id = match self {
             Event::LockFinality(order_id) => order_id,
             Event::CommitFinality(order_id) => order_id,
+            Event::CloseFinality(order_id) => order_id,
             Event::CetTimelockExpired(order_id) => order_id,
             Event::RefundTimelockExpired(order_id) => order_id,
             Event::RefundFinality(order_id) => order_id,

daemon/src/oracle.rs

@@ -78,7 +78,8 @@ impl<CFD, M> Actor<CFD, M> {
                 | CfdState::Open { .. }
                 | CfdState::PendingCommit { .. }
                 | CfdState::OpenCommitted { .. }
-                | CfdState::PendingCet { .. } => {
+                | CfdState::PendingCet { .. } =>
+                {
                     pending_attestations.insert(cfd.order.oracle_event_id);
                 }
 
@@ -88,6 +89,7 @@ impl<CFD, M> Actor<CFD, M> {
                 | CfdState::Accepted { .. }
                 | CfdState::Rejected { .. }
                 | CfdState::ContractSetup { .. }
+                | CfdState::PendingClose { .. }
 
                 // Final states
                 | CfdState::Closed { .. }

daemon/src/taker_cfd.rs

@@ -4,8 +4,7 @@ use crate::db::{
 };
 use crate::model::cfd::{
     Attestation, Cfd, CfdState, CfdStateChangeEvent, CfdStateCommon, Dlc, Order, OrderId, Origin,
-    Role, RollOverProposal, SettlementKind, SettlementProposal, UpdateCfdProposal,
-    UpdateCfdProposals,
+    Role, RollOverProposal, SettlementKind, SettlementProposal, UpdateCfdProposal, UpdateCfdProposals,
 };
 use crate::model::{BitMexPriceEventId, Usd};
 use crate::monitor::{self, MonitorParams};
@@ -356,11 +355,11 @@ impl Actor {
 
         let mut conn = self.db.acquire().await?;
 
-        let cfd = load_cfd_by_order_id(order_id, &mut conn).await?;
+        let mut cfd = load_cfd_by_order_id(order_id, &mut conn).await?;
         let dlc = cfd.open_dlc().context("CFD was in wrong state")?;
 
         let proposal = self.get_settlement_proposal(order_id)?;
-        let (_tx, sig_taker) = dlc.close_transaction(proposal)?;
+        let (tx, sig_taker) = dlc.close_transaction(proposal)?;
 
         self.send_to_maker
             .do_send_async(wire::TakerToMaker::InitiateSettlement {
@@ -369,7 +368,10 @@ impl Actor {
             })
             .await?;
 
-        // TODO: Monitor for the transaction
+        cfd.handle(CfdStateChangeEvent::ProposalSigned(
+            TimestampedTransaction::new(tx),
+        ))?;
+        insert_new_cfd_state_by_order_id(cfd.order.id, cfd.state, &mut conn).await?;
 
         self.remove_pending_proposal(&order_id)?;
 
@@ -547,6 +549,7 @@ impl Actor {
                 },
                 dlc: dlc.clone(),
                 attestation: None,
+                collaborative_close: None,
             },
             &mut conn,
         )

daemon/src/to_sse_event.rs

@@ -133,6 +133,7 @@ pub enum CfdState {
     Open,
     PendingCommit,
     PendingCet,
+    PendingClose,
     OpenCommitted,
     IncomingSettlementProposal,
     OutgoingSettlementProposal,
@@ -329,6 +330,7 @@ fn to_cfd_state(
             model::cfd::CfdState::SetupFailed { .. } => CfdState::SetupFailed,
             model::cfd::CfdState::PendingCommit { .. } => CfdState::PendingCommit,
             model::cfd::CfdState::PendingCet { .. } => CfdState::PendingCet,
+            model::cfd::CfdState::PendingClose { .. } => CfdState::PendingClose,
             model::cfd::CfdState::Closed { .. } => CfdState::Closed,
         },
         Some(UpdateCfdProposal::RollOverProposal {
@@ -386,10 +388,19 @@ fn to_cfd_details(state: model::cfd::CfdState, role: Role, network: Network) ->
             ],
             Some(attestation.payout()),
         ),
-        Closed { attestation, .. } => (
+        Closed {
+            attestation: Some(attestation),
+            ..
+        } => (
             vec![tx_ub.cet(attestation.txid())],
             Some(attestation.payout()),
         ),
+        Closed {
+            attestation: None, ..
+        } => {
+            // TODO: Provide CfdDetails about collaborative settlement
+            (vec![], None)
+        }
         MustRefund { dlc, .. } => (
             vec![tx_ub.lock(&dlc), tx_ub.commit(&dlc), tx_ub.refund(&dlc)],
             Some(dlc.refund_amount(role)),
@@ -397,6 +408,7 @@ fn to_cfd_details(state: model::cfd::CfdState, role: Role, network: Network) ->
         Refunded { dlc, .. } => (vec![tx_ub.refund(&dlc)], Some(dlc.refund_amount(role))),
         OutgoingOrderRequest { .. }
         | IncomingOrderRequest { .. }
+        | PendingClose { .. }
         | Accepted { .. }
         | Rejected { .. }
         | ContractSetup { .. }