refactor: implement RBF withdraw on new client

integrationtests/tests/tests.rs

@@ -26,58 +26,11 @@ use fedimint_core::outcome::TransactionStatus;
 use fedimint_core::task::TaskGroup;
 use fedimint_core::{msats, sats};
 use fedimint_server::epoch::ConsensusItem;
-use fedimint_wallet_server::common::{PegOutFees, Rbf};
 use futures::future::{join_all, Either};
 use serde::{Deserialize, Serialize};
 
 use crate::fixtures::{peers, test};
 
-#[tokio::test(flavor = "multi_thread")]
-async fn wallet_peg_outs_support_rbf() -> Result<()> {
-    test(2, |fed, user, bitcoin| async move {
-        // Need lock to keep tx in mempool from getting mined
-        let bitcoin = bitcoin.lock_exclusive().await;
-        let address = bitcoin.get_new_address().await;
-
-        fed.mine_and_mint(&*user, &*bitcoin, sats(5000)).await;
-        let (fees, out_point) = user.peg_out(1000, &address);
-        fed.run_consensus_epochs(2).await;
-        fed.broadcast_transactions().await;
-
-        let txid = user.await_peg_out_txid(out_point).await.unwrap();
-        assert_eq!(
-            bitcoin.get_mempool_tx_fee(&txid).await,
-            fees.amount().into()
-        );
-
-        // RBF by increasing sats per kvb by 1000
-        let rbf = Rbf {
-            fees: PegOutFees::new(1000, fees.total_weight),
-            txid,
-        };
-        let out_point = user.rbf_peg_out_tx(rbf.clone()).await.unwrap();
-        fed.run_consensus_epochs(2).await;
-        fed.broadcast_transactions().await;
-        let txid = user.await_peg_out_txid(out_point).await.unwrap();
-
-        assert_eq!(
-            bitcoin.get_mempool_tx_fee(&txid).await,
-            (fees.amount() + rbf.fees.amount()).into()
-        );
-
-        assert_eq!(
-            bitcoin.mine_block_and_get_received(&address).await,
-            sats(1000)
-        );
-        bitcoin
-            .mine_blocks(fed.wallet.consensus.finality_delay as u64)
-            .await;
-        fed.run_consensus_epochs(1).await;
-        assert_eq!(fed.max_balance_sheet(), 0);
-    })
-    .await
-}
-
 #[tokio::test(flavor = "multi_thread")]
 async fn wallet_peg_outs_must_wait_for_available_utxos() -> Result<()> {
     test(2, |fed, user, bitcoin| async move {

modules/fedimint-wallet-client/src/lib.rs

@@ -7,7 +7,7 @@ mod withdraw;
 use std::sync::Arc;
 use std::time::SystemTime;
 
-use anyhow::{anyhow, bail, ensure};
+use anyhow::{anyhow, bail, ensure, Context as AnyhowContext};
 use async_stream::stream;
 use bitcoin::{Address, Network};
 use fedimint_bitcoind::{create_bitcoind, DynBitcoindRpc};
@@ -81,6 +81,12 @@ pub trait WalletClientExt {
         fee: PegOutFees,
     ) -> anyhow::Result<OperationId>;
 
+    /// Attempt to increase the fee of a onchain withdraw transaction using
+    /// replace by fee (RBF).
+    /// This can prevent transactions from getting stuck
+    /// in the mempool
+    async fn rbf_withdraw(&self, rbf: Rbf) -> anyhow::Result<OperationId>;
+
     async fn subscribe_withdraw_updates(
         &self,
         operation_id: OperationId,
@@ -176,7 +182,7 @@ impl WalletClientExt for Client {
             .operation_log()
             .get_operation(operation_id)
             .await
-            .ok_or(anyhow!("Operation not found"))?;
+            .with_context(|| anyhow!("Operation not found: {operation_id}"))?;
 
         if operation_log_entry.operation_type() != WalletCommonGen::KIND.as_str() {
             bail!("Operation is not a wallet operation");
@@ -288,6 +294,32 @@ impl WalletClientExt for Client {
         Ok(operation_id)
     }
 
+    async fn rbf_withdraw(&self, rbf: Rbf) -> anyhow::Result<OperationId> {
+        let (wallet_client, instance) =
+            self.get_first_module::<WalletClientModule>(&WalletCommonGen::KIND);
+
+        let operation_id = OperationId(thread_rng().gen());
+
+        let withdraw_output = wallet_client
+            .create_rbf_withdraw_output(operation_id, rbf.clone())
+            .await?;
+        let tx_builder =
+            TransactionBuilder::new().with_output(withdraw_output.into_dyn(instance.id));
+
+        self.finalize_and_submit_transaction(
+            operation_id,
+            WalletCommonGen::KIND.as_str(),
+            move |_, change| WalletOperationMeta::RbfWithdraw {
+                rbf: rbf.clone(),
+                change,
+            },
+            tx_builder,
+        )
+        .await?;
+
+        Ok(operation_id)
+    }
+
     async fn subscribe_withdraw_updates(
         &self,
         operation_id: OperationId,
@@ -299,15 +331,17 @@ impl WalletClientExt for Client {
             .operation_log()
             .get_operation(operation_id)
             .await
-            .ok_or(anyhow!("Operation not found"))?;
+            .with_context(|| anyhow!("Operation not found: {operation_id}"))?;
 
         if operation.operation_type() != WalletCommonGen::KIND.as_str() {
             bail!("Operation is not a wallet operation");
         }
 
         let operation_meta = operation.meta::<WalletOperationMeta>();
 
-        let WalletOperationMeta::Withdraw { change, .. } = operation_meta else {
+        let (WalletOperationMeta::Withdraw { change, .. }
+        | WalletOperationMeta::RbfWithdraw { change, .. }) = operation_meta
+        else {
             bail!("Operation is not a withdraw operation");
         };
 
@@ -436,6 +470,11 @@ pub enum WalletOperationMeta {
         fee: PegOutFees,
         change: Option<OutPoint>,
     },
+
+    RbfWithdraw {
+        rbf: Rbf,
+        change: Option<OutPoint>,
+    },
 }
 
 #[derive(Debug)]
@@ -537,7 +576,7 @@ impl WalletClientModule {
         self.module_api
             .fetch_peg_out_fees(&address, amount)
             .await?
-            .ok_or(anyhow!("Federation didn't return peg-out fees"))
+            .context("Federation didn't return peg-out fees")
     }
 
     pub async fn create_withdraw_output(
@@ -569,6 +608,28 @@ impl WalletClientModule {
             state_machines: Arc::new(sm_gen),
         })
     }
+
+    pub async fn create_rbf_withdraw_output(
+        &self,
+        operation_id: OperationId,
+        rbf: Rbf,
+    ) -> anyhow::Result<ClientOutput<WalletOutput, WalletClientStates>> {
+        let output = WalletOutput::Rbf(rbf);
+
+        let sm_gen = move |txid, out_idx| {
+            vec![WalletClientStates::Withdraw(WithdrawStateMachine {
+                operation_id,
+                state: WithdrawStates::Created(CreatedWithdrawState {
+                    fm_outpoint: OutPoint { txid, out_idx },
+                }),
+            })]
+        };
+
+        Ok(ClientOutput::<WalletOutput, WalletClientStates> {
+            output,
+            state_machines: Arc::new(sm_gen),
+        })
+    }
 }
 
 fn check_address(address: &Address, network: Network) -> anyhow::Result<()> {

modules/fedimint-wallet-tests/tests/tests.rs

@@ -1,5 +1,6 @@
 use std::time::{Duration, SystemTime};
 
+use anyhow::bail;
 use fedimint_client::Client;
 use fedimint_core::util::{BoxStream, NextOrPending};
 use fedimint_core::{sats, Amount, Feerate};
@@ -10,7 +11,7 @@ use fedimint_testing::btc::BitcoinTest;
 use fedimint_testing::fixtures::Fixtures;
 use fedimint_wallet_client::{DepositState, WalletClientExt, WalletClientGen, WithdrawState};
 use fedimint_wallet_common::config::WalletGenParams;
-use fedimint_wallet_common::PegOutFees;
+use fedimint_wallet_common::{PegOutFees, Rbf};
 use fedimint_wallet_server::WalletGen;
 use futures::stream::StreamExt;
 
@@ -87,7 +88,7 @@ async fn on_chain_peg_in_and_peg_out_happy_case() -> anyhow::Result<()> {
     assert_eq!(sub.ok().await?, WithdrawState::Created);
     let txid = match sub.ok().await? {
         WithdrawState::Succeeded(txid) => txid,
-        _ => panic!("Unexpected state"),
+        other => panic!("Unexpected state: {other:?}"),
     };
     bitcoin.get_mempool_tx_fee(&txid).await;
 
@@ -133,3 +134,65 @@ async fn peg_out_fail_refund() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn peg_outs_support_rbf() -> anyhow::Result<()> {
+    let fixtures = fixtures();
+    let fed = fixtures.new_fed().await;
+    let client = fed.new_client().await;
+    let bitcoin = fixtures.bitcoin();
+    // Need lock to keep tx in mempool from getting mined
+    let bitcoin = bitcoin.lock_exclusive().await;
+
+    let finality_delay = 10;
+    bitcoin.mine_blocks(finality_delay).await;
+
+    let mut balance_sub = peg_in(&client, bitcoin.as_ref(), finality_delay).await?;
+
+    let address = bitcoin.get_new_address().await;
+    let peg_out = bsats(PEG_OUT_AMOUNT_SATS);
+    let fees = client.get_withdraw_fee(address.clone(), peg_out).await?;
+    let op = client.withdraw(address.clone(), peg_out, fees).await?;
+
+    let sub = client.subscribe_withdraw_updates(op).await?;
+    let mut sub = sub.into_stream();
+    assert_eq!(sub.ok().await?, WithdrawState::Created);
+    let state = sub.ok().await?;
+    let WithdrawState::Succeeded(txid) = state else {
+        bail!("Unexpected state: {state:?}")
+    };
+    assert_eq!(
+        bitcoin.get_mempool_tx_fee(&txid).await,
+        fees.amount().into()
+    );
+    let balance_after_normal_peg_out =
+        sats(PEG_IN_AMOUNT_SATS - PEG_OUT_AMOUNT_SATS - fees.amount().to_sat());
+    assert_eq!(client.get_balance().await, balance_after_normal_peg_out);
+    assert_eq!(balance_sub.ok().await?, balance_after_normal_peg_out);
+
+    // RBF by increasing sats per kvb by 1000
+    let rbf = Rbf {
+        fees: PegOutFees::new(1000, fees.total_weight),
+        txid,
+    };
+    let op = client.rbf_withdraw(rbf.clone()).await?;
+    let sub = client.subscribe_withdraw_updates(op).await?;
+    let mut sub = sub.into_stream();
+    assert_eq!(sub.ok().await?, WithdrawState::Created);
+    let txid = match sub.ok().await? {
+        WithdrawState::Succeeded(txid) => txid,
+        other => panic!("Unexpected state: {other:?}"),
+    };
+    let total_fees = fees.amount() + rbf.fees.amount();
+    assert_eq!(bitcoin.get_mempool_tx_fee(&txid).await, total_fees.into());
+    assert_eq!(
+        bitcoin.mine_block_and_get_received(&address).await,
+        sats(PEG_OUT_AMOUNT_SATS)
+    );
+
+    let balance_after_rbf_peg_out =
+        sats(PEG_IN_AMOUNT_SATS - PEG_OUT_AMOUNT_SATS - total_fees.to_sat());
+    assert_eq!(client.get_balance().await, balance_after_rbf_peg_out);
+    assert_eq!(balance_sub.ok().await?, balance_after_rbf_peg_out);
+    Ok(())
+}