darkfi follower implementation

Reviewed-on: https://codeberg.org/darkrenaissance/swapd/pulls/10

Cargo.lock

@@ -1525,6 +1525,7 @@ dependencies = [
 [[package]]
 name = "darkfi"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "arti-client",
  "async-trait",
@@ -1573,6 +1574,7 @@ dependencies = [
 [[package]]
 name = "darkfi-derive"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "darkfi-derive-internal",
  "proc-macro-crate 3.1.0",
@@ -1584,6 +1586,7 @@ dependencies = [
 [[package]]
 name = "darkfi-derive-internal"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1593,6 +1596,7 @@ dependencies = [
 [[package]]
 name = "darkfi-sdk"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "blake2b_simd",
  "blake3",
@@ -1614,6 +1618,7 @@ dependencies = [
 [[package]]
 name = "darkfi-serial"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "async-trait",
  "blake3",
@@ -1628,6 +1633,7 @@ dependencies = [
 [[package]]
 name = "darkfi_dao_contract"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "blake2b_simd",
  "bs58",
@@ -1646,6 +1652,7 @@ dependencies = [
 [[package]]
 name = "darkfi_deployooor_contract"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "darkfi",
  "darkfi-sdk",
@@ -1660,6 +1667,7 @@ dependencies = [
 [[package]]
 name = "darkfi_money_contract"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "blake3",
  "bs58",
@@ -1997,6 +2005,7 @@ checksum = "9ea835d29036a4087793836fa931b08837ad5e957da9e23886b29586fb9b6650"
 [[package]]
 name = "drk"
 version = "0.4.1"
+source = "git+https://github.com/noot/darkfi?branch=drk-updates#eb6d51660b781e0420c426500f5f24d64e636ef1"
 dependencies = [
  "blake3",
  "bs58",
@@ -6185,6 +6194,7 @@ dependencies = [
  "easy-parallel",
  "ethers",
  "log",
+ "pasta_curves",
  "serde",
  "signal-hook",
  "signal-hook-async-std",

Cargo.toml

@@ -35,13 +35,14 @@ serde = {version = "1.0.197", features = ["derive"]}
 structopt = "0.3.26"
 structopt-toml = "0.5.1"
 
+async-std = {version = "1.12.0", features = ["attributes", "tokio1"]}
 async-watch = "0.3.1"
 ethers = { version = "2.0", features = ["ethers-solc", "ws"] }
+pasta_curves = "0.5.1"
 thiserror = "1.0.57"
 
 [dev-dependencies]
 swapd = { path = ".", features = ["test-utils"] }
-async-std = {version = "1.12.0", features = ["attributes", "tokio1"]}
 
 [features]
 default = []
@@ -52,11 +53,12 @@ halo2_proofs = {git="https://github.com/parazyd/halo2", branch="v4"}
 halo2_gadgets = {git="https://github.com/parazyd/halo2", branch="v4"}
 
 [patch."https://github.com/darkrenaissance/darkfi"]
-darkfi = { path = "/home/e/darkfi" }
-darkfi-serial = { path = "/home/e/darkfi/src/serial" }
-darkfi-sdk = { path = "/home/e/darkfi/src/sdk" }
-drk = { path = "/home/e/darkfi/bin/drk" }
-
-#darkfi = { git = "https://github.com/noot/darkfi", branch = "fix-include-bytes" }
-#darkfi-serial = { git = "https://github.com/noot/darkfi", branch = "fix-include-bytes" }
-#drk = { git = "https://github.com/noot/darkfi", branch = "fix-include-bytes" }
+#darkfi = { path = "/home/e/darkfi" }
+#darkfi-serial = { path = "/home/e/darkfi/src/serial" }
+#darkfi-sdk = { path = "/home/e/darkfi/src/sdk" }
+#drk = { path = "/home/e/darkfi/bin/drk" }
+
+darkfi = { git = "https://github.com/noot/darkfi", branch = "drk-updates" }
+darkfi-serial = { git = "https://github.com/noot/darkfi", branch = "drk-updates" }
+darkfi-sdk = { git = "https://github.com/noot/darkfi", branch = "drk-updates" }
+drk = { git = "https://github.com/noot/darkfi", branch = "drk-updates" }

ethereum/src/SwapCreator.sol

@@ -78,10 +78,12 @@ contract SwapCreator is Secp256k1 {
         bytes32 swapID,
         bytes32 claimKey,
         bytes32 refundKey,
+        address claimer,
         uint256 timeout1,
         uint256 timeout2,
         address asset,
-        uint256 value
+        uint256 value,
+        uint256 nonce
     );
     event Ready(bytes32 indexed swapID);
     event Claimed(bytes32 indexed swapID, bytes32 indexed s);
@@ -168,7 +170,7 @@ contract SwapCreator is Secp256k1 {
         bytes32 _claimCommitment,
         bytes32 _refundCommitment,
         address payable _claimer,
-        uint256 _timeoutDuration1,
+        uint256 _timeoutDuration1,  
         uint256 _timeoutDuration2,
         address _asset,
         uint256 _value,
@@ -208,10 +210,12 @@ contract SwapCreator is Secp256k1 {
             swapID,
             _claimCommitment,
             _refundCommitment,
+            _claimer,
             swap.timeout1,
             swap.timeout2,
             swap.asset,
-            swap.value
+            swap.value,
+            swap.nonce
         );
         swaps[swapID] = Stage.PENDING;
         return swapID;

src/darkfi/error.rs

@@ -1,25 +1,35 @@
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
-    #[error("failed to create drk struct: {0}")]
+    #[error("failed to create drk struct")]
     DrkInitializationFailed(#[source] darkfi::Error),
-    #[error("failed to initialize wallet: {0}")]
+    #[error("failed to initialize wallet")]
     InitializeWallet(#[source] darkfi::Error),
-    #[error("failed to initialize money: {0}")]
+    #[error("failed to initialize money")]
     InitializeMoney(#[source] drk::WalletDbError),
-    #[error("failed to initialize dao: {0}")]
+    #[error("failed to initialize dao")]
     InitializeDao(#[source] drk::WalletDbError),
-    #[error("failed to get default secret: {0}")]
+    #[error("failed to get default secret")]
     DefaultSecret(#[source] darkfi::Error),
-    #[error("failed to generate keypair: {0}")]
+    #[error("failed to generate keypair")]
     Keygen(#[source] drk::WalletDbError),
-    #[error("failed to get money balance: {0}")]
+    #[error("failed to get money balance")]
     MoneyBalance(#[source] darkfi::Error),
-    #[error("failed to get aliases mapped by token: {0}")]
+    #[error("failed to get aliases mapped by token")]
     GetAliasesMappedByToken(#[source] darkfi::Error),
-    #[error("failed to build transfer: {0}")]
+    #[error("failed to build transfer")]
     BuildTransfer(#[source] darkfi::Error),
-    #[error("failed to submit transaction: {0}")]
+    #[error("failed to submit transaction")]
     SubmitTransaction(#[source] darkfi::Error),
     #[error("swap keypair does not exist; generate one first")]
     SwapKeypairDoesNotExist,
+    #[error("failed to put keypair")]
+    PutKeypair(#[source] drk::WalletDbError),
+    #[error("failed convert counterparty secret bytes to secret key")]
+    CounterpartySecretKeyConversionFailed(#[source] darkfi_sdk::ContractError),
+    #[error("listening to Ready event stream failed")]
+    ReadyEventStreamFailed,
+    #[error("listening to Refunded event stream failed")]
+    RefundedEventStreamFailed,
+    #[error("middleware error: {0}")]
+    MiddlewareError(String),
 }

src/darkfi/follower.rs

@@ -0,0 +1,91 @@
+use crate::protocol::traits::FollowerArgs;
+use darkfi_sdk::crypto::{Keypair, SecretKey};
+
+use super::{wallet::Wallet, Error};
+use crate::protocol::traits::{ContractSwapArgs, Follower};
+
+/// Implemented on top of the non-initiating chain
+///
+/// Can probably become an extension trait of an RPC client eventually
+pub(crate) trait OtherChainClient {
+    fn claim_funds(
+        &self,
+        our_secret: [u8; 32],
+        contract_swap_args: ContractSwapArgs,
+    ) -> Result<(), crate::Error>;
+}
+
+pub(crate) struct DrkFollower<C: OtherChainClient> {
+    other_chain_client: C,
+    secret: SecretKey,
+    wallet: Wallet,
+    args: FollowerArgs,
+    contract_swap_args: Option<ContractSwapArgs>,
+}
+
+impl<C: OtherChainClient> DrkFollower<C> {
+    fn new(other_chain_client: C, secret: SecretKey, wallet: Wallet, args: FollowerArgs) -> Self {
+        Self { other_chain_client, secret, wallet, args, contract_swap_args: None }
+    }
+}
+
+impl<C: OtherChainClient + Send + Sync> Follower for DrkFollower<C> {
+    // handle the swap initiation by locking funds on chain B
+    fn handle_counterparty_funds_locked(
+        &mut self,
+        contract_swap_args: ContractSwapArgs,
+    ) -> Result<(), crate::Error> {
+        self.contract_swap_args = Some(contract_swap_args);
+
+        // lock DRK funds to shared swap account
+        let shared_swap_public_key = Keypair::new(self.secret).public;
+
+        // cursed hack b/c `exec_sql` takes `dyn ToSql` which is not Send or Sync :/
+        let tx = async_std::task::block_on(
+            self.wallet.build_swap_transfer(self.args.value, shared_swap_public_key),
+        )?;
+        let _tx_hash = async_std::task::block_on(self.wallet.submit_transaction(&tx))?;
+        Ok(())
+    }
+
+    // handle the funds being ready to be claimed by us
+    fn handle_ready_to_claim(&self) -> Result<(), crate::Error> {
+        use darkfi_sdk::crypto::pasta_prelude::PrimeField as _;
+
+        let our_secret: [u8; 32] = self
+            .secret
+            .inner()
+            .to_repr()
+            .as_ref()
+            .try_into()
+            .expect("can convert secret key to 32 byte representation");
+        self.other_chain_client.claim_funds(
+            our_secret,
+            self.contract_swap_args.as_ref().expect("`contract_swap_id` must be set").clone(),
+        )?;
+        Ok(())
+    }
+
+    // handle the counterparty refunding their funds, in case of a timeout
+    fn handle_counterparty_funds_refunded(
+        &self,
+        counterparty_secret: [u8; 32],
+    ) -> Result<(), crate::Error> {
+        let counterparty_secret_key = SecretKey::from_bytes(counterparty_secret)
+            .map_err(|e| crate::Error::from(Error::CounterpartySecretKeyConversionFailed(e)))?;
+        let counterparty_secret_fp = counterparty_secret_key.inner();
+        let our_secret_fp = self.secret.inner();
+        let shared_secret_fp = counterparty_secret_fp + our_secret_fp;
+        let shared_secret = SecretKey::from(shared_secret_fp);
+        let keypair = Keypair::new(shared_secret);
+
+        async_std::task::block_on(self.wallet.put_keypair(keypair))?;
+        let tx = async_std::task::block_on(self.wallet.build_transfer(
+            self.args.value,
+            self.wallet.user_keypair().public,
+            keypair,
+        ))?;
+        let _tx_hash = async_std::task::block_on(self.wallet.submit_transaction(&tx))?;
+        Ok(())
+    }
+}

src/darkfi/follower_event_watcher.rs

@@ -0,0 +1,110 @@
+use crate::{
+    darkfi::Error,
+    ethereum::{swap_creator::SwapCreator, utils},
+    protocol::{follower::Event, traits::FollowerEventWatcher},
+};
+use ethers::prelude::Middleware;
+use smol::{channel, stream::StreamExt as _};
+use std::sync::Arc;
+
+pub(crate) struct Watcher;
+
+#[darkfi_serial::async_trait]
+impl FollowerEventWatcher for Watcher {
+    async fn run_counterparty_funds_locked_watcher<M: Middleware>(
+        event_tx: channel::Sender<Event>,
+        contract: SwapCreator<M>,
+        middleware: Arc<M>,
+        claim_commitment: [u8; 32],
+        refund_commitment: [u8; 32],
+        from_block: u64,
+    ) -> Result<(), crate::Error> {
+        // watch for a `NewSwap` event with the correct swap parameters
+        // note: we still need to check for correct asset, value, and timeout.
+        let topic2: ethers::types::U256 = claim_commitment.into();
+        let topic3: ethers::types::U256 = refund_commitment.into();
+        let events = contract
+            .ready_filter()
+            .from_block(from_block)
+            .address(contract.address().into())
+            .topic2(topic2) // `newSwap` event sig is topic0 and `contract_swap_id` is topic1
+            .topic3(topic3);
+
+        // TODO: ensure that the funds locked have the correct parameters before locking:
+        // - value
+        // - asset
+        // - timeouts
+
+        let mut stream = events.stream().await.unwrap().with_meta();
+
+        // we listen for the first event, as there can only be one event
+        // that matches the filter (ie. has the same swap_id)
+        let Some(Ok((_event, meta))) = stream.next().await else {
+            return Err(Error::ReadyEventStreamFailed.into());
+        };
+
+        let receipt = middleware
+            .get_transaction_receipt(meta.transaction_hash)
+            .await
+            .map_err(|e| {
+                Error::MiddlewareError(format!("failed to get transaction receipt: {:?}", e))
+            })?
+            .expect("receipt must exist if log exists");
+
+        let (contract_swap, _) = utils::parse_new_swap_event_from_receipt(receipt)?;
+
+        event_tx.send(Event::CounterpartyFundsLocked(contract_swap)).await.unwrap();
+        Ok(())
+    }
+
+    async fn run_ready_to_claim_watcher<M: Middleware>(
+        event_tx: channel::Sender<Event>,
+        contract: SwapCreator<M>,
+        contract_swap_id: &[u8; 32],
+        from_block: u64,
+    ) -> Result<(), crate::Error> {
+        let topic1: ethers::types::U256 = contract_swap_id.into();
+        let events = contract
+            .ready_filter()
+            .from_block(from_block)
+            .address(contract.address().into())
+            .topic1(topic1); // `ready` event sig is topic0
+
+        let mut stream = events.stream().await.unwrap().with_meta();
+
+        // we listen for the first event, as there can only be one event
+        // that matches the filter (ie. has the same swap_id)
+        let Some(Ok((_, _meta))) = stream.next().await else {
+            return Err(Error::ReadyEventStreamFailed.into());
+        };
+
+        event_tx.send(Event::ReadyToClaim).await.unwrap();
+        Ok(())
+    }
+
+    async fn run_counterparty_funds_refunded_watcher<M: Middleware>(
+        event_tx: channel::Sender<Event>,
+        contract: SwapCreator<M>,
+        contract_swap_id: &[u8; 32],
+        from_block: u64,
+    ) -> Result<(), crate::Error> {
+        let topic1: ethers::types::U256 = contract_swap_id.into();
+        let events = contract
+            .refunded_filter()
+            .from_block(from_block)
+            .address(contract.address().into())
+            .topic1(topic1); // `refunded` event sig is topic0
+
+        let mut stream = events.stream().await.unwrap().with_meta();
+
+        // we listen for the first event, as there can only be one event
+        // that matches the filter (ie. has the same swap_id)
+        let Some(Ok((event, _meta))) = stream.next().await else {
+            return Err(Error::RefundedEventStreamFailed.into());
+        };
+
+        let counterparty_secret = event.s;
+        event_tx.send(Event::CounterpartyFundsRefunded(counterparty_secret)).await.unwrap();
+        Ok(())
+    }
+}

src/darkfi/mod.rs

@@ -1,4 +1,6 @@
 mod error;
+mod follower;
+mod follower_event_watcher;
 pub(crate) mod wallet;
 
 pub(crate) use error::Error;

src/darkfi/wallet.rs

@@ -26,7 +26,7 @@ impl Wallet {
         endpoint: Url,
         ex: Arc<smol::Executor<'static>>,
     ) -> Result<Self, crate::Error> {
-        let drk = Drk::new(wallet_path, wallet_pass, endpoint, ex)
+        let drk = Drk::new(wallet_path, wallet_pass, Some(endpoint), ex)
             .await
             .map_err(|e| crate::Error::from(Error::DrkInitializationFailed(e)))?;
         drk.initialize_wallet()
@@ -42,12 +42,20 @@ impl Wallet {
         Ok(Self { drk, user_keypair, swap_keypair: None })
     }
 
+    pub(crate) fn user_keypair(&self) -> Keypair {
+        self.user_keypair
+    }
+
     pub(crate) async fn generate_swap_keypair(&mut self) -> Result<(), crate::Error> {
         self.swap_keypair =
             Some(self.drk.money_keygen().await.map_err(|e| crate::Error::from(Error::Keygen(e)))?);
         Ok(())
     }
 
+    pub(crate) async fn put_keypair(&self, keypair: Keypair) -> Result<(), crate::Error> {
+        self.drk.put_keypair(keypair).await.map_err(|e| crate::Error::from(Error::PutKeypair(e)))
+    }
+
     // get the user's DRK balance
     pub(crate) async fn get_user_balance(&self) -> Result<u64, crate::Error> {
         get_balance(&self.drk, self.user_keypair.secret).await
@@ -91,6 +99,20 @@ impl Wallet {
             .map_err(|e| crate::Error::from(Error::BuildTransfer(e)))
     }
 
+    pub(crate) async fn build_transfer(
+        &self,
+        amount: u128,
+        recipient: PublicKey,
+        signer: Keypair,
+    ) -> Result<Transaction, crate::Error> {
+        let token_id =
+            self.drk.get_token("DRK".to_string()).await.expect("token id must exist for DRK");
+        self.drk
+            .transfer_with_signer(&amount.to_string(), token_id, recipient, signer)
+            .await
+            .map_err(|e| crate::Error::from(Error::BuildTransfer(e)))
+    }
+
     // submit a transaction to the network
     pub(crate) async fn submit_transaction(
         &self,

src/ethereum/error.rs

@@ -34,4 +34,8 @@ pub enum Error {
     ExpectedFixedBytes(ethers::abi::Token),
     #[error("expected two U256s, got something else")]
     ExpectedTwoU256s,
+    #[error("expected address, got another token type: {0}")]
+    ExpectedAddress(ethers::abi::Token),
+    #[error("expected U256, got another token type: {0}")]
+    ExpectedU256(ethers::abi::Token),
 }

src/ethereum/initiator.rs

@@ -1,16 +1,12 @@
 use crate::{
-    ethereum::{
-        swap_creator::{Swap, SwapCreator},
-        Error,
-    },
-    protocol::traits::{HandleCounterpartyKeysReceivedResult, InitiateSwapArgs, Initiator},
+    ethereum::{swap_creator::SwapCreator, utils, Error},
+    protocol::traits::{ContractSwapArgs, HandleCounterpartyKeysReceivedResult, Initiator},
 };
 
 use darkfi_serial::async_trait;
 use ethers::prelude::*;
 
 use log::info;
-use std::sync::Arc;
 
 /// Implemented on top of the non-initiating chain
 ///
@@ -25,20 +21,14 @@ pub(crate) trait OtherChainClient {
 
 pub(crate) struct EthInitiator<M: Middleware, C: OtherChainClient> {
     contract: SwapCreator<M>,
-    middleware: Arc<M>,
     other_chain_client: C,
     secret: [u8; 32],
 }
 
 #[allow(dead_code)]
 impl<M: Middleware, C: OtherChainClient> EthInitiator<M, C> {
-    pub(crate) fn new(
-        contract: SwapCreator<M>,
-        middleware: Arc<M>,
-        other_chain_client: C,
-        secret: [u8; 32],
-    ) -> Self {
-        Self { contract, middleware, other_chain_client, secret }
+    pub(crate) fn new(contract: SwapCreator<M>, other_chain_client: C, secret: [u8; 32]) -> Self {
+        Self { contract, other_chain_client, secret }
     }
 }
 
@@ -46,19 +36,18 @@ impl<M: Middleware, C: OtherChainClient> EthInitiator<M, C> {
 impl<M: Middleware + 'static, C: OtherChainClient + Send + Sync> Initiator for EthInitiator<M, C> {
     async fn handle_counterparty_keys_received(
         &self,
-        args: InitiateSwapArgs,
+        args: ContractSwapArgs,
     ) -> Result<HandleCounterpartyKeysReceivedResult, crate::Error> {
-        use ethers::abi::ParamType;
-
-        let InitiateSwapArgs {
+        let ContractSwapArgs {
             claim_commitment,
             refund_commitment,
             claimer,
-            timeout_duration_1,
-            timeout_duration_2,
+            timeout_1,
+            timeout_2,
             asset,
             value,
             nonce,
+            ..
         } = args;
 
         // TODO: ERC20 is *not* handled right now
@@ -72,8 +61,8 @@ impl<M: Middleware + 'static, C: OtherChainClient + Send + Sync> Initiator for E
                 claim_commitment,
                 refund_commitment,
                 claimer,
-                timeout_duration_1,
-                timeout_duration_2,
+                timeout_1,
+                timeout_2,
                 asset,
                 value,
                 nonce,
@@ -87,74 +76,9 @@ impl<M: Middleware + 'static, C: OtherChainClient + Send + Sync> Initiator for E
             .map_err(|e| Error::FailedToAwaitPendingTransaction("new_swap".to_string(), e))?
             .ok_or_else(|| Error::NoReceipt)?;
 
-        if receipt.status != Some(U64::from(1)) {
-            return Err(Error::TransactionFailed("new_swap".to_string(), receipt).into());
-        }
-
-        if receipt.logs.len() != 1 {
-            return Err(Error::NewSwapUnexpectedLogCount(receipt.logs.len()).into());
-        }
-
-        if receipt.logs[0].topics.len() != 1 {
-            return Err(Error::NewSwapUnexpectedTopicCount(receipt.logs[0].topics.len()).into());
-        }
-
-        let log_data = &receipt.logs[0].data;
-
-        // ABI-unpack log data
-        // note: there are other parameters emitted in the log, but we don't care about them
-        let mut tokens = ethers::abi::decode(
-            &vec![
-                ParamType::FixedBytes(32),
-                ParamType::FixedBytes(32),
-                ParamType::FixedBytes(32),
-                ParamType::Uint(256),
-                ParamType::Uint(256),
-            ],
-            &log_data.0,
-        )
-        .map_err(|e| Error::NewSwapLogDecodingFailed(e))?;
-
-        if tokens.len() != 5 {
-            return Err(Error::NewSwapUnexpectedLogTokenCount(tokens.len()).into());
-        }
-
-        let swap_id = match tokens.remove(0) {
-            ethers::abi::Token::FixedBytes(bytes) => {
-                // this shouldn't happen, would be an error in ethers-rs
-                if bytes.len() != 32 {
-                    return Err(Error::FixedBytesDecodingError(bytes.len()).into());
-                }
-
-                let mut swap_id = [0u8; 32];
-                swap_id.copy_from_slice(&bytes);
-                swap_id
-            }
-            token => {
-                return Err(Error::ExpectedFixedBytes(token).into());
-            }
-        };
-        let (timeout_1, timeout_2) = match (tokens.remove(2), tokens.remove(2)) {
-            // tokens index 3 and 4
-            (ethers::abi::Token::Uint(timeout_1), ethers::abi::Token::Uint(timeout_2)) => {
-                (timeout_1, timeout_2)
-            }
-            _ => {
-                return Err(Error::ExpectedTwoU256s.into());
-            }
-        };
-
-        let contract_swap = Swap {
-            owner: self.middleware.default_sender().expect("must have a default sender"),
-            claim_commitment,
-            refund_commitment,
-            claimer,
-            timeout_1,
-            timeout_2,
-            asset,
-            value,
-            nonce,
-        };
+        let block_number =
+            receipt.block_number.expect("block number must be set in receipt").as_u64();
+        let (contract_swap, swap_id) = utils::parse_new_swap_event_from_receipt(receipt)?;
 
         info!(
             "initiated swap on-chain: contract_swap_id = {}",
@@ -164,10 +88,7 @@ impl<M: Middleware + 'static, C: OtherChainClient + Send + Sync> Initiator for E
         Ok(HandleCounterpartyKeysReceivedResult {
             contract_swap_id: swap_id,
             contract_swap,
-            block_number: receipt
-                .block_number
-                .expect("block number must be set in receipt")
-                .as_u64(),
+            block_number,
         })
     }
 

src/ethereum/initiator_event_watcher.rs

@@ -42,10 +42,10 @@ impl InitiatorEventWatcher for Watcher {
     ) -> Result<(), crate::Error> {
         let topic1: ethers::types::U256 = contract_swap_id.into();
         let events = contract
-            .claimed_filter() // claimed event sig is topic0
+            .claimed_filter()
             .from_block(from_block)
             .address(contract.address().into())
-            .topic1(topic1);
+            .topic1(topic1); // `claimed` event sig is topic0
 
         let mut stream = events.stream().await.unwrap().with_meta();
 

src/ethereum/mod.rs

@@ -2,6 +2,7 @@ mod error;
 pub(crate) mod initiator;
 mod initiator_event_watcher;
 pub(crate) mod swap_creator;
+pub(crate) mod utils;
 
 pub use error::Error;
 #[allow(unused_imports)]

src/ethereum/swap_creator.rs

@@ -1,3 +1,3 @@
 use ethers::prelude::*;
 
-abigen!(SwapCreator, "./ethereum/out/SwapCreator.sol/SwapCreator.json",);
+abigen!(SwapCreator, "./ethereum/out/SwapCreator.sol/SwapCreator.json");

src/ethereum/utils.rs

@@ -0,0 +1,143 @@
+use crate::ethereum::{error::Error, swap_creator::Swap};
+use ethers::{
+    abi::{Address, ParamType},
+    types::{TransactionReceipt, U64},
+};
+
+/// Parse a `newSwap` event from a transaction receipt.
+///
+/// Returns the `Swap` struct and the swap ID.
+pub(crate) fn parse_new_swap_event_from_receipt(
+    receipt: TransactionReceipt,
+) -> Result<(Swap, [u8; 32]), crate::Error> {
+    if receipt.status != Some(U64::from(1)) {
+        return Err(Error::TransactionFailed("new_swap".to_string(), receipt).into());
+    }
+
+    if receipt.logs.len() != 1 {
+        return Err(Error::NewSwapUnexpectedLogCount(receipt.logs.len()).into());
+    }
+
+    if receipt.logs[0].topics.len() != 1 {
+        return Err(Error::NewSwapUnexpectedTopicCount(receipt.logs[0].topics.len()).into());
+    }
+
+    let log_data = &receipt.logs[0].data;
+
+    // ABI-unpack log data
+    // note: there are other parameters emitted in the log, but we don't care about them
+    let mut tokens = ethers::abi::decode(
+        &vec![
+            ParamType::FixedBytes(32), // swapID
+            ParamType::FixedBytes(32), // claimKey
+            ParamType::FixedBytes(32), // refundKey
+            ParamType::Address,        // claimer
+            ParamType::Uint(256),      // timeout_1
+            ParamType::Uint(256),      // timeout_2
+            ParamType::Address,        // asset
+            ParamType::Uint(256),      // value
+            ParamType::Uint(256),      // nonce
+        ],
+        &log_data.0,
+    )
+    .map_err(|e| Error::NewSwapLogDecodingFailed(e))?;
+
+    if tokens.len() != 9 {
+        return Err(Error::NewSwapUnexpectedLogTokenCount(tokens.len()).into());
+    }
+
+    let swap_id = match tokens.remove(0) {
+        ethers::abi::Token::FixedBytes(bytes) => {
+            // this shouldn't happen, would be an error in ethers-rs
+            if bytes.len() != 32 {
+                return Err(Error::FixedBytesDecodingError(bytes.len()).into());
+            }
+
+            let mut swap_id = [0u8; 32];
+            swap_id.copy_from_slice(&bytes);
+            swap_id
+        }
+        token => {
+            return Err(Error::ExpectedFixedBytes(token).into());
+        }
+    };
+    let claim_commitment = match tokens.remove(0) {
+        ethers::abi::Token::FixedBytes(bytes) => {
+            if bytes.len() != 32 {
+                return Err(Error::FixedBytesDecodingError(bytes.len()).into());
+            }
+
+            let mut claim_commitment = [0u8; 32];
+            claim_commitment.copy_from_slice(&bytes);
+            claim_commitment
+        }
+        token => {
+            return Err(Error::ExpectedFixedBytes(token).into());
+        }
+    };
+    let refund_commitment = match tokens.remove(0) {
+        ethers::abi::Token::FixedBytes(bytes) => {
+            if bytes.len() != 32 {
+                return Err(Error::FixedBytesDecodingError(bytes.len()).into());
+            }
+
+            let mut refund_commitment = [0u8; 32];
+            refund_commitment.copy_from_slice(&bytes);
+            refund_commitment
+        }
+        token => {
+            return Err(Error::ExpectedFixedBytes(token).into());
+        }
+    };
+
+    let claimer: Address = match tokens.remove(0) {
+        ethers::abi::Token::Address(bytes) => bytes.into(),
+        token => {
+            return Err(Error::ExpectedAddress(token).into());
+        }
+    };
+
+    let (timeout_1, timeout_2) = match (tokens.remove(0), tokens.remove(0)) {
+        // tokens index 3 and 4
+        (ethers::abi::Token::Uint(timeout_1), ethers::abi::Token::Uint(timeout_2)) => {
+            (timeout_1, timeout_2)
+        }
+        _ => {
+            return Err(Error::ExpectedTwoU256s.into());
+        }
+    };
+
+    let asset: Address = match tokens.remove(0) {
+        ethers::abi::Token::Address(bytes) => bytes.into(),
+        token => {
+            return Err(Error::ExpectedAddress(token).into());
+        }
+    };
+
+    let value = match tokens.remove(0) {
+        ethers::abi::Token::Uint(value) => value,
+        token => {
+            return Err(Error::ExpectedU256(token).into());
+        }
+    };
+
+    let nonce = match tokens.remove(0) {
+        ethers::abi::Token::Uint(nonce) => nonce,
+        token => {
+            return Err(Error::ExpectedU256(token).into());
+        }
+    };
+
+    let contract_swap = Swap {
+        owner: receipt.from,
+        claim_commitment,
+        refund_commitment,
+        claimer,
+        timeout_1,
+        timeout_2,
+        asset,
+        value,
+        nonce,
+    };
+    Ok((contract_swap, swap_id))
+}

src/protocol/error.rs

@@ -2,6 +2,7 @@ use crate::protocol::traits::CounterpartyKeys;
 
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
+    // initiator errors
     #[error("unexpected received counterparty keys event: {0}")]
     UnexpectedReceivedCounterpartyKeysEvent(CounterpartyKeys),
     #[error("unexpected counterparty funds locked event")]
@@ -12,4 +13,12 @@ pub enum Error {
     UnexpectedAlmostTimeout1Event,
     #[error("unexpected past timeout 2 event")]
     UnexpectedPastTimeout2Event,
+
+    // follower errors
+    #[error("unexpected counterparty funds locked event")]
+    UnexpectedCounterpartyFundsLocked,
+    #[error("unexpected ready to claim event")]
+    UnexpectedReadyToClaim,
+    #[error("unexpected counterparty funds refunded event")]
+    UnexpectedCounterpartyFundsRefunded,
 }

src/protocol/follower.rs

@@ -1,41 +1,109 @@
+use super::{traits::ContractSwapArgs, Error};
 use crate::protocol::traits::Follower;
 use smol::channel;
 
+use log::{info, warn};
+
 #[allow(dead_code)]
-enum Event {
-    CounterpartyFundsLocked,
+pub(crate) enum Event {
+    // occurs when the counterparty has locked funds in the contract.
+    // contains the swap id within the contract.
+    CounterpartyFundsLocked(ContractSwapArgs),
     ReadyToClaim,
-    CounterpartyFundsRefunded,
+    CounterpartyFundsRefunded([u8; 32]),
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum State {
+    WaitingForCounterpartyFundsLocked,
+    WaitingForContractReady,
+    Completed,
 }
 
 #[allow(dead_code)]
 struct Swap {
-    handler: Box<dyn Follower>,
+    // the chain-specific event handler
+    handler: Box<dyn Follower + Send + Sync>,
+
+    // the event receiver channel for the swap
+    // the [`Watcher`] sends events to this channel
     event_rx: channel::Receiver<Event>,
+
+    // the current state of the swap
+    state_tx: async_watch::Sender<State>,
+    state_rx: async_watch::Receiver<State>,
 }
 
 #[allow(dead_code)]
 impl Swap {
-    fn new(handler: Box<dyn Follower>, event_rx: channel::Receiver<Event>) -> Self {
-        Self { handler, event_rx }
+    fn new(
+        handler: Box<dyn Follower + Send + Sync>,
+        event_rx: channel::Receiver<Event>,
+    ) -> (Self, async_watch::Receiver<State>) {
+        let state = async_watch::channel(State::WaitingForCounterpartyFundsLocked);
+        (Self { handler, event_rx, state_tx: state.0, state_rx: state.1.clone() }, state.1)
     }
 
-    async fn run(&mut self) {
+    async fn run(&mut self) -> Result<(), crate::Error> {
         loop {
             match self.event_rx.recv().await {
-                Ok(Event::CounterpartyFundsLocked) => {
-                    self.handler.handle_counterparty_funds_locked();
+                Ok(Event::CounterpartyFundsLocked(contract_swap_args)) => {
+                    info!("counterparty funds locked");
+
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForCounterpartyFundsLocked)
+                    {
+                        warn!(
+                            "unexpected event CounterpartyFundsLocked, state is {:?}",
+                            *self.state_rx.borrow()
+                        );
+                        return Err(Error::UnexpectedCounterpartyFundsLocked.into());
+                    }
+
+                    self.handler.handle_counterparty_funds_locked(contract_swap_args)?;
+
+                    self.state_tx
+                        .send(State::WaitingForContractReady)
+                        .expect("state channel should not be dropped");
                 }
                 Ok(Event::ReadyToClaim) => {
-                    self.handler.handle_ready_to_claim();
+                    info!("ready to claim funds on counterparty chain");
+
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForContractReady) {
+                        warn!(
+                            "unexpected event ReadyToClaim, state is {:?}",
+                            *self.state_rx.borrow()
+                        );
+                        return Err(Error::UnexpectedReadyToClaim.into());
+                    }
+
+                    self.handler.handle_ready_to_claim()?;
+                    self.state_tx
+                        .send(State::Completed)
+                        .expect("state channel should not be dropped");
                 }
-                Ok(Event::CounterpartyFundsRefunded) => {
-                    self.handler.handle_counterparty_funds_refunded();
+                Ok(Event::CounterpartyFundsRefunded(counterparty_secret)) => {
+                    info!("counterparty refunded funds");
+                    if !matches!(*self.state_rx.borrow(), State::WaitingForContractReady) {
+                        warn!(
+                            "unexpected event CounterpartyFundsRefunded, state is {:?}",
+                            *self.state_rx.borrow()
+                        );
+                        return Err(Error::UnexpectedCounterpartyFundsRefunded.into());
+                    }
+
+                    self.handler.handle_counterparty_funds_refunded(counterparty_secret)?;
+                    self.state_tx
+                        .send(State::Completed)
+                        .expect("state channel should not be dropped");
                 }
                 Err(_) => {
+                    info!("event channel closed, exiting");
                     break;
                 }
             }
         }
+
+        Ok(())
     }
 }

src/protocol/initiator.rs

@@ -1,6 +1,6 @@
 use crate::protocol::{
     traits::{
-        CounterpartyKeys, HandleCounterpartyKeysReceivedResult, InitiateSwapArgs, InitiationArgs,
+        ContractSwapArgs, CounterpartyKeys, HandleCounterpartyKeysReceivedResult, InitiationArgs,
         Initiator,
     },
     Error,
@@ -32,7 +32,7 @@ struct Swap {
     args: InitiationArgs,
 
     // the chain-specific event handler
-    // TODO: just make this a generic
+    // TODO: just make this a generic?
     handler: Box<dyn Initiator + Send + Sync>,
 
     // the event receiver channel for the swap
@@ -96,12 +96,13 @@ impl Swap {
                     let refund_commitment =
                         ethers::utils::keccak256(&counterparty_keys.secp256k1_public_key);
 
-                    let args = InitiateSwapArgs {
+                    let args = ContractSwapArgs {
+                        owner: self.args.owner,
+                        claimer: self.args.claimer,
                         claim_commitment: self.args.claim_commitment,
                         refund_commitment,
-                        claimer: self.args.claimer,
-                        timeout_duration_1: self.args.timeout_duration_1,
-                        timeout_duration_2: self.args.timeout_duration_2,
+                        timeout_1: self.args.timeout_duration_1,
+                        timeout_2: self.args.timeout_duration_2,
                         asset: self.args.asset,
                         value: self.args.value,
                         nonce: self.args.nonce,
@@ -206,6 +207,7 @@ impl Swap {
                 }
             }
 
+            // TODO: handle the case where the swap is *not* completed
             if matches!(*self.state_rx.borrow(), State::Completed) {
                 info!("swap completed, exiting");
                 break;
@@ -219,7 +221,6 @@ impl Swap {
 #[cfg(test)]
 mod test {
     use super::*;
-
     use std::sync::Arc;
 
     use smol::channel::bounded;
@@ -258,8 +259,7 @@ mod test {
 
         let other_chain_client = MockOtherChainClient;
         let refund_secret = [0; 32]; // TODO generate an actual secp256k1 private key for refund testing
-        let initiator =
-            EthInitiator::new(contract.clone(), signer.clone(), other_chain_client, refund_secret);
+        let initiator = EthInitiator::new(contract.clone(), other_chain_client, refund_secret);
 
         // TODO: this is the same key as the initiator right now.
         let counterparty_secret: [u8; 32] = anvil.keys()[0].to_bytes().try_into().unwrap();
@@ -269,6 +269,7 @@ mod test {
         let claim_commitment = ethers::utils::keccak256(pubkey_bytes);
 
         let args = InitiationArgs {
+            owner: signer.address(),
             claim_commitment,
             claimer: signer.address(),
             timeout_duration_1: U256::from(120),
@@ -294,9 +295,10 @@ mod test {
             .send(CounterpartyKeys { secp256k1_public_key: [0; 33] })
             .await
             .expect("should send counterparty keys");
+        smol::future::block_on(join_handle)
+            .expect("run_received_counterparty_keys_watcher should finish");
         state.changed().await.expect("state should change");
         assert!(*state.borrow() == State::WaitingForCounterpartyFundsLocked);
-        join_handle.cancel().await.unwrap().unwrap();
 
         Watcher::run_counterparty_funds_locked_watcher(event_tx.clone())
             .await

src/protocol/mod.rs

@@ -1,6 +1,6 @@
 //! This module contains the protocol traits and logic for DRK-ETH atomic swaps.
 mod error;
-mod follower;
+pub(crate) mod follower;
 pub(crate) mod initiator;
 pub(crate) mod traits;
 

src/protocol/traits.rs

@@ -1,20 +1,28 @@
 use crate::ethereum::swap_creator::Swap; // TODO: shouldn't depend on this
-use crate::{error::Error, ethereum::swap_creator::SwapCreator, protocol::initiator::Event};
+use crate::{
+    error::Error,
+    ethereum::swap_creator::SwapCreator,
+    protocol::{follower, initiator},
+};
 use darkfi_serial::async_trait;
 use ethers::{prelude::*, utils::hex};
+use pasta_curves::pallas;
 use smol::channel;
 use std::{
     fmt,
     fmt::{Display, Formatter},
 };
 
+pub(crate) use crate::ethereum::swap_creator::Swap as ContractSwapArgs;
+
 // Initial parameters required by the swap initiator.
 // TODO: make Address/U256 generic; these are ethers-specific right now
 #[allow(dead_code)]
 #[derive(Debug, Clone)]
 pub(crate) struct InitiationArgs {
-    pub(crate) claim_commitment: [u8; 32],
+    pub(crate) owner: Address,
     pub(crate) claimer: Address,
+    pub(crate) claim_commitment: [u8; 32],
     pub(crate) timeout_duration_1: U256,
     pub(crate) timeout_duration_2: U256,
     pub(crate) asset: Address,
@@ -22,17 +30,11 @@ pub(crate) struct InitiationArgs {
     pub(crate) nonce: U256,
 }
 
-// TODO: make Address/U256 generic; these are ethers-specific right now
+// Initial parameters required by the swap follower.
 #[derive(Debug)]
-pub(crate) struct InitiateSwapArgs {
-    pub(crate) claim_commitment: [u8; 32],
-    pub(crate) refund_commitment: [u8; 32],
-    pub(crate) claimer: Address,
-    pub(crate) timeout_duration_1: U256,
-    pub(crate) timeout_duration_2: U256,
-    pub(crate) asset: Address,
-    pub(crate) value: U256,
-    pub(crate) nonce: U256,
+pub(crate) struct FollowerArgs {
+    pub(crate) counterparty_public_key: pallas::Point, // TODO: make this generic
+    pub(crate) value: u128,                            // TODO: is u128 sufficient?
 }
 
 // TODO: make this generic for both chains
@@ -76,7 +78,7 @@ pub(crate) trait Initiator {
     // initiates the swap by locking funds on chain A
     async fn handle_counterparty_keys_received(
         &self,
-        args: InitiateSwapArgs,
+        args: ContractSwapArgs,
     ) -> Result<HandleCounterpartyKeysReceivedResult, Error>;
 
     // handles the counterparty locking funds
@@ -99,30 +101,30 @@ pub(crate) trait Initiator {
 #[async_trait]
 pub(crate) trait InitiatorEventWatcher {
     async fn run_received_counterparty_keys_watcher(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
         counterparty_keys_rx: channel::Receiver<CounterpartyKeys>,
     ) -> Result<(), Error>;
 
     async fn run_counterparty_funds_locked_watcher(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
     ) -> Result<(), Error>;
 
     // TODO: make this generic for both chains
     async fn run_counterparty_funds_claimed_watcher<M: Middleware>(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
         contract: SwapCreator<M>,
         contract_swap_id: &[u8; 32],
         from_block: u64,
     ) -> Result<(), Error>;
 
     async fn run_timeout_1_watcher(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
         timeout_1: u64,
         buffer_seconds: u64,
     ) -> Result<(), Error>;
 
     async fn run_timeout_2_watcher(
-        event_tx: channel::Sender<Event>,
+        event_tx: channel::Sender<initiator::Event>,
         timeout_2: u64,
     ) -> Result<(), Error>;
 }
@@ -130,11 +132,43 @@ pub(crate) trait InitiatorEventWatcher {
 /// the chain that is the counterparty to the swap; ie. the second-mover
 pub(crate) trait Follower {
     // handle the swap initiation by locking funds on chain B
-    fn handle_counterparty_funds_locked(&self);
+    fn handle_counterparty_funds_locked(
+        &mut self,
+        contract_swap_id: ContractSwapArgs,
+    ) -> Result<(), crate::Error>;
 
     // handle the funds being ready to be claimed by us
-    fn handle_ready_to_claim(&self);
+    fn handle_ready_to_claim(&self) -> Result<(), crate::Error>;
 
     // handle the counterparty refunding their funds, in case of a timeout
-    fn handle_counterparty_funds_refunded(&self);
+    fn handle_counterparty_funds_refunded(
+        &self,
+        counterparty_secret: [u8; 32],
+    ) -> Result<(), crate::Error>;
+}
+
+#[async_trait]
+pub(crate) trait FollowerEventWatcher {
+    async fn run_counterparty_funds_locked_watcher<M: Middleware>(
+        event_tx: channel::Sender<follower::Event>,
+        contract: SwapCreator<M>,
+        middleware: std::sync::Arc<M>,
+        claim_commitment: [u8; 32],
+        refund_commitment: [u8; 32],
+        from_block: u64,
+    ) -> Result<(), Error>;
+
+    async fn run_ready_to_claim_watcher<M: Middleware>(
+        event_tx: channel::Sender<follower::Event>,
+        contract: SwapCreator<M>,
+        contract_swap_id: &[u8; 32],
+        from_block: u64,
+    ) -> Result<(), Error>;
+
+    async fn run_counterparty_funds_refunded_watcher<M: Middleware>(
+        event_tx: channel::Sender<follower::Event>,
+        contract: SwapCreator<M>,
+        contract_swap_id: &[u8; 32],
+        from_block: u64,
+    ) -> Result<(), Error>;
 }