Successful publication of lock tx

Cargo.lock

@@ -436,6 +436,7 @@ dependencies = [
  "sha2",
  "sqlx",
  "tempfile",
+ "thiserror",
  "tokio",
  "tokio-util",
  "uuid",

daemon/Cargo.toml

@@ -21,6 +21,7 @@ serde_json = "1"
 serde_with = { version = "1", features = ["macros"] }
 sha2 = "0.9"
 sqlx = { version = "0.5", features = ["offline"] }
+thiserror = "1"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "net"] }
 tokio-util = { version = "0.6", features = ["codec"] }
 uuid = { version = "0.8", features = ["serde", "v4"] }

daemon/src/maker.rs

@@ -76,7 +76,7 @@ async fn main() -> Result<()> {
         ext_priv_key,
     )
     .await?;
-    let wallet_info = wallet.sync().unwrap();
+    let wallet_info = wallet.sync().await.unwrap();
 
     let oracle = schnorrsig::KeyPair::new(SECP256K1, &mut rand::thread_rng()); // TODO: Fetch oracle public key from oracle.
 

daemon/src/maker_cfd_actor.rs

@@ -70,7 +70,7 @@ pub fn new(
             while let Some(message) = receiver.recv().await {
                 match message {
                     maker_cfd_actor::Command::SyncWallet => {
-                        let wallet_info = wallet.sync().unwrap();
+                        let wallet_info = wallet.sync().await.unwrap();
                         wallet_feed_sender.send(wallet_info).unwrap();
                     }
                     maker_cfd_actor::Command::TakeOrder {
@@ -175,7 +175,7 @@ pub fn new(
                         let cfd = load_cfd_by_order_id(order_id, &mut conn).await.unwrap();
                         let margin = cfd.margin().unwrap();
 
-                        let maker_params = wallet.build_party_params(margin, pk).unwrap();
+                        let maker_params = wallet.build_party_params(margin, pk).await.unwrap();
 
                         let (actor, inbox) = setup_contract_actor::new(
                             {
@@ -195,6 +195,7 @@ pub fn new(
                             sk,
                             oracle_pk,
                             cfd,
+                            wallet.clone(),
                         );
 
                         current_contract_setup = Some(inbox.clone());
@@ -255,7 +256,7 @@ pub fn new(
                                 common: CfdStateCommon {
                                     transition_timestamp: SystemTime::now(),
                                 },
-                                dlc,
+                                dlc: dlc.clone(),
                             },
                             &mut conn,
                         )
@@ -266,9 +267,12 @@ pub fn new(
                             .send(load_all_cfds(&mut conn).await.unwrap())
                             .unwrap();
 
-                        // TODO: Publish on chain and only then transition to open - this might
-                        // require saving some internal state to make sure we are able to monitor
-                        // the publication after a restart
+                        let txid = wallet.try_broadcast_transaction(dlc.lock).await.unwrap();
+
+                        println!("Lock transaction published with txid {}", txid);
+
+                        // TODO: tx monitoring, once confirmed with x blocks transition the Cfd to
+                        // Open
                     }
                 }
             }

daemon/src/model/cfd.rs

@@ -1,7 +1,6 @@
 use crate::model::{Leverage, Position, TradingPair, Usd};
 use anyhow::Result;
 use bdk::bitcoin::secp256k1::{SecretKey, Signature};
-use bdk::bitcoin::util::psbt::PartiallySignedTransaction;
 use bdk::bitcoin::{Amount, Transaction};
 use cfd_protocol::secp256k1_zkp::EcdsaAdaptorSignature;
 use rust_decimal::Decimal;
@@ -511,7 +510,8 @@ pub struct Dlc {
     pub revocation: SecretKey,
     pub publish: SecretKey,
 
-    pub lock: PartiallySignedTransaction,
+    /// The fully signed lock transaction ready to be published on chain
+    pub lock: Transaction,
     pub commit: (Transaction, EcdsaAdaptorSignature),
     pub cets: Vec<(Transaction, EcdsaAdaptorSignature, RangeInclusive<u64>)>,
     pub refund: (Transaction, Signature),

daemon/src/setup_contract_actor.rs

@@ -1,5 +1,6 @@
 use crate::model::cfd::{Cfd, Dlc};
-use crate::wire::{Msg0, Msg1, SetupMsg};
+use crate::wallet::Wallet;
+use crate::wire::{Msg0, Msg1, Msg2, SetupMsg};
 use anyhow::{Context, Result};
 use bdk::bitcoin::secp256k1::{schnorrsig, SecretKey, Signature, SECP256K1};
 use bdk::bitcoin::{Amount, PublicKey, Transaction};
@@ -26,6 +27,7 @@ pub fn new(
     sk: SecretKey,
     oracle_pk: schnorrsig::PublicKey,
     cfd: Cfd,
+    wallet: Wallet,
 ) -> (impl Future<Output = Dlc>, mpsc::UnboundedSender<SetupMsg>) {
     let (sender, mut receiver) = mpsc::unbounded_channel::<SetupMsg>();
 
@@ -127,11 +129,22 @@ pub fn new(
             .map(|(tx, _, digits)| (digits.range(), (tx, digits)))
             .collect::<HashMap<_, _>>();
 
+        let mut signed_lock_tx = wallet.sign(lock_tx).await.unwrap();
+        send_to_other(SetupMsg::Msg2(Msg2 {
+            signed_lock: signed_lock_tx.clone(),
+        }));
+        let msg2 = receiver.recv().await.unwrap().try_into_msg2().unwrap();
+        signed_lock_tx.merge(msg2.signed_lock).unwrap();
+
+        // TODO: In case we sign+send but never receive (the signed lock_tx from the other party) we
+        // need some fallback handling (after x time) to spend the outputs in a different way so the
+        // other party cannot hold us hostage
+
         Dlc {
             identity: sk,
             revocation: rev_sk,
             publish: publish_sk,
-            lock: lock_tx,
+            lock: signed_lock_tx.extract_tx(),
             commit: (commit_tx, msg1.commit),
             cets: msg1
                 .cets

daemon/src/taker.rs

@@ -80,7 +80,7 @@ async fn main() -> Result<()> {
         ext_priv_key,
     )
     .await?;
-    let wallet_info = wallet.sync().unwrap();
+    let wallet_info = wallet.sync().await.unwrap();
 
     let oracle = schnorrsig::KeyPair::new(SECP256K1, &mut rand::thread_rng()); // TODO: Fetch oracle public key from oracle.
 

daemon/src/taker_cfd_actor.rs

@@ -49,7 +49,7 @@ pub fn new(
             while let Some(message) = receiver.recv().await {
                 match message {
                     Command::SyncWallet => {
-                        let wallet_info = wallet.sync().unwrap();
+                        let wallet_info = wallet.sync().await.unwrap();
                         wallet_feed_sender.send(wallet_info).unwrap();
                     }
                     Command::TakeOrder { order_id, quantity } => {
@@ -114,7 +114,7 @@ pub fn new(
                         let cfd = load_cfd_by_order_id(order_id, &mut conn).await.unwrap();
                         let margin = cfd.margin().unwrap();
 
-                        let taker_params = wallet.build_party_params(margin, pk).unwrap();
+                        let taker_params = wallet.build_party_params(margin, pk).await.unwrap();
 
                         let (actor, inbox) = setup_contract_actor::new(
                             {
@@ -125,6 +125,7 @@ pub fn new(
                             sk,
                             oracle_pk,
                             cfd,
+                            wallet.clone(),
                         );
 
                         tokio::spawn({
@@ -160,7 +161,7 @@ pub fn new(
                                 common: CfdStateCommon {
                                     transition_timestamp: SystemTime::now(),
                                 },
-                                dlc,
+                                dlc: dlc.clone(),
                             },
                             &mut conn,
                         )
@@ -171,11 +172,12 @@ pub fn new(
                             .send(load_all_cfds(&mut conn).await.unwrap())
                             .unwrap();
 
-                        // TODO: Some code duplication with maker in this block
+                        let txid = wallet.try_broadcast_transaction(dlc.lock).await.unwrap();
 
-                        // TODO: Publish on chain and only then transition to open - this might
-                        // require saving some internal state to make sure we are able to monitor
-                        // the publication after a restart
+                        println!("Lock transaction published with txid {}", txid);
+
+                        // TODO: tx monitoring, once confirmed with x blocks transition the Cfd to
+                        // Open
                     }
                 }
             }

daemon/src/wallet.rs

@@ -1,20 +1,29 @@
 use crate::model::WalletInfo;
-use anyhow::{Context, Result};
+use anyhow::{anyhow, bail, Context, Result};
 use bdk::bitcoin::util::bip32::ExtendedPrivKey;
-use bdk::bitcoin::{Amount, PublicKey};
+use bdk::bitcoin::util::psbt::PartiallySignedTransaction;
+use bdk::bitcoin::{Amount, PublicKey, Transaction, Txid};
 use bdk::blockchain::{ElectrumBlockchain, NoopProgress};
 use bdk::wallet::AddressIndex;
-use bdk::KeychainKind;
+use bdk::{electrum_client, Error, KeychainKind, SignOptions};
 use cfd_protocol::{PartyParams, WalletExt};
+use rocket::serde::json::Value;
 use std::path::Path;
+use std::sync::Arc;
 use std::time::SystemTime;
+use tokio::sync::Mutex;
 
 const SLED_TREE_NAME: &str = "wallet";
 
-pub struct Wallet<B = ElectrumBlockchain, D = bdk::sled::Tree> {
-    wallet: bdk::Wallet<B, D>,
+#[derive(Clone)]
+pub struct Wallet {
+    wallet: Arc<Mutex<bdk::Wallet<ElectrumBlockchain, bdk::sled::Tree>>>,
 }
 
+#[derive(thiserror::Error, Debug, Clone, Copy)]
+#[error("The transaction is already in the blockchain")]
+pub struct TransactionAlreadyInBlockchain;
+
 impl Wallet {
     pub async fn new(
         electrum_rpc_url: &str,
@@ -35,23 +44,27 @@ impl Wallet {
             ElectrumBlockchain::from(client),
         )?;
 
+        let wallet = Arc::new(Mutex::new(wallet));
+
         Ok(Self { wallet })
     }
 
-    pub fn build_party_params(
+    pub async fn build_party_params(
         &self,
         amount: Amount,
         identity_pk: PublicKey,
     ) -> Result<PartyParams> {
-        self.wallet.build_party_params(amount, identity_pk)
+        let wallet = self.wallet.lock().await;
+        wallet.build_party_params(amount, identity_pk)
     }
 
-    pub fn sync(&self) -> Result<WalletInfo> {
-        self.wallet.sync(NoopProgress, None)?;
+    pub async fn sync(&self) -> Result<WalletInfo> {
+        let wallet = self.wallet.lock().await;
+        wallet.sync(NoopProgress, None)?;
 
-        let balance = self.wallet.get_balance()?;
+        let balance = wallet.get_balance()?;
 
-        let address = self.wallet.get_address(AddressIndex::LastUnused)?.address;
+        let address = wallet.get_address(AddressIndex::LastUnused)?.address;
 
         let wallet_info = WalletInfo {
             balance: Amount::from_sat(balance),
@@ -61,4 +74,86 @@ impl Wallet {
 
         Ok(wallet_info)
     }
+
+    pub async fn sign(
+        &self,
+        mut psbt: PartiallySignedTransaction,
+    ) -> Result<PartiallySignedTransaction> {
+        let wallet = self.wallet.lock().await;
+
+        wallet
+            .sign(
+                &mut psbt,
+                SignOptions {
+                    trust_witness_utxo: true,
+                    ..Default::default()
+                },
+            )
+            .context("could not sign transaction")?;
+
+        Ok(psbt)
+    }
+
+    pub async fn try_broadcast_transaction(&self, tx: Transaction) -> Result<Txid> {
+        let wallet = self.wallet.lock().await;
+        // TODO: Optimize this match to be a map_err / more readable in general
+        let txid = tx.txid();
+        match wallet.broadcast(tx) {
+            Ok(txid) => Ok(txid),
+            Err(e) => {
+                if let Error::Electrum(electrum_client::Error::Protocol(ref value)) = e {
+                    let error_code = match parse_rpc_protocol_error_code(value) {
+                        Ok(error_code) => error_code,
+                        Err(inner) => {
+                            eprintln!("Failed to parse error code from RPC message: {}", inner);
+                            return Err(anyhow!(e));
+                        }
+                    };
+
+                    if error_code == i64::from(RpcErrorCode::RpcVerifyAlreadyInChain) {
+                        return Ok(txid);
+                    }
+                }
+
+                Err(anyhow!(e))
+            }
+        }
+    }
+}
+
+fn parse_rpc_protocol_error_code(error_value: &Value) -> anyhow::Result<i64> {
+    let json_map = match error_value {
+        serde_json::Value::Object(map) => map,
+        _ => bail!("Json error is not json object "),
+    };
+
+    let error_code_value = match json_map.get("code") {
+        Some(val) => val,
+        None => bail!("No error code field"),
+    };
+
+    let error_code_number = match error_code_value {
+        serde_json::Value::Number(num) => num,
+        _ => bail!("Error code is not a number"),
+    };
+
+    if let Some(int) = error_code_number.as_i64() {
+        Ok(int)
+    } else {
+        bail!("Error code is not an unsigned integer")
+    }
+}
+
+/// Bitcoin error codes: https://github.com/bitcoin/bitcoin/blob/97d3500601c1d28642347d014a6de1e38f53ae4e/src/rpc/protocol.h#L23
+pub enum RpcErrorCode {
+    /// Transaction or block was rejected by network rules. Error code -27.
+    RpcVerifyAlreadyInChain,
+}
+
+impl From<RpcErrorCode> for i64 {
+    fn from(code: RpcErrorCode) -> Self {
+        match code {
+            RpcErrorCode::RpcVerifyAlreadyInChain => -27,
+        }
+    }
 }

daemon/src/wire.rs

@@ -34,8 +34,23 @@ pub enum MakerToTaker {
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(tag = "type", content = "payload")]
 pub enum SetupMsg {
+    /// Message enabling setting up lock and based on that commit, refund and cets
+    ///
+    /// Each party sends and receives this message.
+    /// After receiving this message each party is able to construct the lock transaction.
     Msg0(Msg0),
+    /// Message that ensures complete commit, cets and refund transactions
+    ///
+    /// Each party sends and receives this message.
+    /// After receiving this message the commit, refund and cet transactions are complete.
+    /// Once verified we can sign and send the lock PSBT.
     Msg1(Msg1),
+    /// Message adding signature to the lock PSBT
+    ///
+    /// Each party sends and receives this message.
+    /// Upon receiving this message from the other party we merge our signature and then the lock
+    /// tx is fully signed and can be published on chain.
+    Msg2(Msg2),
 }
 
 impl SetupMsg {
@@ -54,6 +69,14 @@ impl SetupMsg {
             Err(self)
         }
     }
+
+    pub fn try_into_msg2(self) -> Result<Msg2, Self> {
+        if let Self::Msg2(v) = self {
+            Ok(v)
+        } else {
+            Err(self)
+        }
+    }
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -137,3 +160,8 @@ impl From<CfdTransactions> for Msg1 {
         }
     }
 }
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Msg2 {
+    pub signed_lock: PartiallySignedTransaction, // TODO: Use binary representation
+}