Merge #230: Refactor sync_kyoto_client

974c8d5 refactor(payjoin): add payjoin error variants to BDKCliError (Mshehu5)
35d8313 refactor(payjoin): implement polling-based monitoring with timeout (Mshehu5)
b88426a refactor: use BlockchainClient as references (Mshehu5)
99b71fd refactor: use handle pattern for Kyoto client (Mshehu5)

Pull request description:

  <!-- You can erase any parts of this template not applicable to your Pull Request. -->

  ### Description

  <!-- Describe the purpose of this PR, what's being adding and/or fixed -->
  This PR addresses issues encountered while implementing persistence for Payjoin specifically around the BlockchainClient only being an owned variable rather than being able to be borrowed/referenced as &Blockchainclient.
  While working on persistence I ran into problems while working on resume command which needs a blockchain client to resume states such as monitor_payjoin_proposal (receiver) and process_payjoin_proposal (sender)
  Because BlockchainClient can only be owned the current design will require a function signature of passing two separate clients to resume sender and receiver states. I initially considered splitting the command into resume_send and resume_receive but this does not fully solve the issue. In particular the sender’s process_payjoin_proposal may call broadcast_transaction and potentially broadcast multiple transactions for persisted send entries stored in the database which still requires reusable access to the client.

  This Ownership issue was previously mentioned in #200 and is also noted in a comment at the top of monitor_payjoin_proposal. It prevents the function from being able to resync multiple times and reliably detect when a transaction appears in the mempool. The root cause is that the Kyoto client Box<LightClient> is destructured and spawned into other tasks when passed through sync_kyoto_client making it unusable afterward.

  What this PR changes
  This PR fixes the issue by refactoring sync_kyoto_client

  - The logic responsible for running the Kyoto node and logger is moved into new_blockchain_client. This makes it that node is started at start of command and not during every sync

  - Instead of returning a Box<lightClient> the function now returns a KyotoClientHandle. Previously the boxed client takes ownership when destructured inside sync_kyoto_client, preventing reuse/reference.
  With the new design sync_kyoto_client takes &KyotoClientHandle, allowing the client to be Refrenced which can be used  for syncing and broadcasting transactions without being owned

  - Additionally monitor_payjoin_proposal is refactored to support resyncing demonstrating that the Kyoto client refactor successfully resolves the original limitations

  ### Notes to the reviewers

  <!-- In this section you can include notes directed to the reviewers, like explaining why some parts
  of the PR were done in a specific way -->
  After refactor I tested the kyoto client on regtest (Cause I do not have access to a signet) I had to set a trusted peer in the code to connect with a cbf count of 1 this worked and I also made transaction using the steps below:
  N.B Payjoin was also tested for the monitor_payjoin_proposal refactor using steps in project readme
  ```
  bitcoin.conf

  regtest=1
  server=1
  rpcuser=user
  rpcpassword=password
  rpcallowip=127.0.0.1
  blockfilterindex=1
  listen=1
  fallbackfee=0.001

  [regtest]
  bind=127.0.0.1
  port=18444
  peerblockfilters=1
  ```

  Step 1: Create transaction
  ```
  PSBT=$(cargo run --features cbf,sqlite -- \
    --network $NETWORK \
    wallet \
    --wallet sender_wallet \
    --ext-descriptor "$SENDER_EXT_DESC" \
    --int-descriptor "$SENDER_INT_DESC" \
    --database-type $DATABASE_TYPE \
    create_tx --to $RECEIVER_ADDR:50000 --fee_rate 1.0 | jq -r '.psbt')
  ```

  Step 2: Sign transaction
  ```
  SIGNED_PSBT=$(cargo run --features cbf,sqlite -- \
    --network $NETWORK \
    wallet \
    --wallet sender_wallet \
    --ext-descriptor "$SENDER_EXT_DESC" \
    --int-descriptor "$SENDER_INT_DESC" \
    --database-type $DATABASE_TYPE \
    sign "$PSBT" | jq -r '.psbt')
  ```

  Step 3: Broadcast transaction
  ```
  cargo run --features cbf,sqlite -- \
    --network $NETWORK \
    wallet \
    --wallet sender_wallet \
    --ext-descriptor "$SENDER_EXT_DESC" \
    --int-descriptor "$SENDER_INT_DESC" \
    --database-type $DATABASE_TYPE \
    --client-type $CLIENT_TYPE \
    --cbf-peer $CBF_PEER \
    --cbf-conn-count $CBF_CONN_COUNT \
    broadcast --psbt "$SIGNED_PSBT"
  ```
   Mine a block to confirm
  `bitcoin-cli -regtest generatetoaddress 1 $(bitcoin-cli -regtest getnewaddress)`

  Checking Transaction Status
  After broadcasting, wait a moment and sync your wallet:
  Sync wallet
  ```
  cargo run --features cbf,sqlite -- \
    --network $NETWORK \
    wallet \
    --wallet receiver_wallet \
    --ext-descriptor "$RECEIVER_EXT_DESC" \
    --int-descriptor "$RECEIVER_INT_DESC" \
    --database-type $DATABASE_TYPE \
    --client-type $CLIENT_TYPE \
    --cbf-peer $CBF_PEER \
    --cbf-conn-count $CBF_CONN_COUNT \
    sync
  ```

  Check balance
  ```
  cargo run --features cbf,sqlite -- \
    --network $NETWORK \
    wallet \
    --wallet receiver_wallet \
    --ext-descriptor "$RECEIVER_EXT_DESC" \
    --int-descriptor "$RECEIVER_INT_DESC" \
    --database-type $DATABASE_TYPE \
    balance
  ```

  List recent transactions
  ```
  cargo run --features cbf,sqlite -- \
    --network $NETWORK \
    wallet \
    --wallet sender_wallet \
    --ext-descriptor "$SENDER_EXT_DESC" \
    --int-descriptor "$SENDER_INT_DESC" \
    --database-type $DATABASE_TYPE \
    transactions
  ```

  ### Checklists

  #### All Submissions:

  * [x] I've signed all my commits
  * [x] I followed the [contribution guidelines](https://github.com/bitcoindevkit/bdk-cli/blob/master/CONTRIBUTING.md)
  * [x] I ran `cargo fmt` and `cargo clippy` before committing

ACKs for top commit:
  tvpeter:
    ACK 974c8d5
  notmandatory:
    ACK 974c8d5

Tree-SHA512: 823eefdde1900fd03fc698466bce414bdba0aceae428d6fbb5af7ef0a0962a0e16948d58badc24b6f99ff9498727c912131d29f46468b421cd54d663e3ec0e8b

Author: tvpeter

src/error.rs

@@ -112,6 +112,34 @@ pub enum BDKCliError {
     ))]
     #[error("Reqwest error: {0}")]
     ReqwestError(#[from] reqwest::Error),
+
+    #[cfg(feature = "payjoin")]
+    #[error("Payjoin URL parse error: {0}")]
+    PayjoinUrlParse(#[from] payjoin::IntoUrlError),
+
+    #[cfg(feature = "payjoin")]
+    #[error("Payjoin send response error: {0}")]
+    PayjoinSendResponse(#[from] payjoin::send::ResponseError),
+
+    #[cfg(feature = "payjoin")]
+    #[error("Payjoin sender build error: {0}")]
+    PayjoinSenderBuild(#[from] payjoin::send::BuildSenderError),
+
+    #[cfg(feature = "payjoin")]
+    #[error("Payjoin receive error: {0}")]
+    PayjoinReceive(#[from] payjoin::receive::Error),
+
+    #[cfg(feature = "payjoin")]
+    #[error("Payjoin selection error: {0}")]
+    PayjoinSelection(#[from] payjoin::receive::SelectionError),
+
+    #[cfg(feature = "payjoin")]
+    #[error("Payjoin input contribution error: {0}")]
+    PayjoinInputContribution(#[from] payjoin::receive::InputContributionError),
+
+    #[cfg(feature = "payjoin")]
+    #[error("Payjoin create request error: {0}")]
+    PayjoinCreateRequest(#[from] payjoin::send::v2::CreateRequestError),
 }
 
 impl From<ExtractTxError> for BDKCliError {

src/handlers.rs

@@ -15,6 +15,8 @@ use crate::config::{WalletConfig, WalletConfigInner};
 use crate::error::BDKCliError as Error;
 #[cfg(any(feature = "sqlite", feature = "redb"))]
 use crate::persister::Persister;
+#[cfg(feature = "cbf")]
+use crate::utils::BlockchainClient::KyotoClient;
 use crate::utils::*;
 #[cfg(feature = "redb")]
 use bdk_redb::Store as RedbStore;
@@ -46,8 +48,6 @@ use bdk_wallet::{
 };
 use cli_table::{Cell, CellStruct, Style, Table, format::Justify};
 use serde_json::json;
-#[cfg(feature = "cbf")]
-use {crate::utils::BlockchainClient::KyotoClient, bdk_kyoto::LightClient, tokio::select};
 
 #[cfg(feature = "electrum")]
 use crate::utils::BlockchainClient::Electrum;
@@ -605,7 +605,7 @@ pub fn handle_offline_wallet_subcommand(
 ))]
 pub(crate) async fn handle_online_wallet_subcommand(
     wallet: &mut Wallet,
-    client: BlockchainClient,
+    client: &BlockchainClient,
     online_subcommand: OnlineWalletSubCommand,
 ) -> Result<String, Error> {
     match online_subcommand {
@@ -632,7 +632,7 @@ pub(crate) async fn handle_online_wallet_subcommand(
                     client
                         .populate_tx_cache(wallet.tx_graph().full_txs().map(|tx_node| tx_node.tx));
 
-                    let update = client.full_scan(request, _stop_gap, batch_size, false)?;
+                    let update = client.full_scan(request, _stop_gap, *batch_size, false)?;
                     wallet.apply_update(update)?;
                 }
                 #[cfg(feature = "esplora")]
@@ -641,7 +641,7 @@ pub(crate) async fn handle_online_wallet_subcommand(
                     parallel_requests,
                 } => {
                     let update = client
-                        .full_scan(request, _stop_gap, parallel_requests)
+                        .full_scan(request, _stop_gap, *parallel_requests)
                         .await
                         .map_err(|e| *e)?;
                     wallet.apply_update(update)?;
@@ -658,7 +658,7 @@ pub(crate) async fn handle_online_wallet_subcommand(
                         hash: genesis_block.block_hash(),
                     });
                     let mut emitter = Emitter::new(
-                        &*client,
+                        client.as_ref(),
                         genesis_cp.clone(),
                         genesis_cp.height(),
                         NO_EXPECTED_MEMPOOL_TXS,
@@ -1246,7 +1246,7 @@ pub(crate) async fn handle_command(cli_opts: CliOpts) -> Result<String, Error> {
 
                 let result = handle_online_wallet_subcommand(
                     &mut wallet,
-                    blockchain_client,
+                    &blockchain_client,
                     online_subcommand,
                 )
                 .await?;
@@ -1258,7 +1258,7 @@ pub(crate) async fn handle_command(cli_opts: CliOpts) -> Result<String, Error> {
                 let mut wallet = new_wallet(network, wallet_opts)?;
                 let blockchain_client =
                     crate::utils::new_blockchain_client(wallet_opts, &wallet, database_path)?;
-                handle_online_wallet_subcommand(&mut wallet, blockchain_client, online_subcommand)
+                handle_online_wallet_subcommand(&mut wallet, &blockchain_client, online_subcommand)
                     .await?
             };
             Ok(result)
@@ -1452,7 +1452,7 @@ async fn respond(
         } => {
             let blockchain =
                 new_blockchain_client(wallet_opts, wallet, _datadir).map_err(|e| e.to_string())?;
-            let value = handle_online_wallet_subcommand(wallet, blockchain, online_subcommand)
+            let value = handle_online_wallet_subcommand(wallet, &blockchain, online_subcommand)
                 .await
                 .map_err(|e| e.to_string())?;
             Some(value)
@@ -1508,7 +1508,7 @@ async fn respond(
     feature = "rpc"
 ))]
 /// Syncs a given wallet using the blockchain client.
-pub async fn sync_wallet(client: BlockchainClient, wallet: &mut Wallet) -> Result<(), Error> {
+pub async fn sync_wallet(client: &BlockchainClient, wallet: &mut Wallet) -> Result<(), Error> {
     #[cfg(any(feature = "electrum", feature = "esplora"))]
     let request = wallet
         .start_sync_with_revealed_spks()
@@ -1523,7 +1523,7 @@ pub async fn sync_wallet(client: BlockchainClient, wallet: &mut Wallet) -> Resul
             // already have.
             client.populate_tx_cache(wallet.tx_graph().full_txs().map(|tx_node| tx_node.tx));
 
-            let update = client.sync(request, batch_size, false)?;
+            let update = client.sync(request, *batch_size, false)?;
             wallet
                 .apply_update(update)
                 .map_err(|e| Error::Generic(e.to_string()))
@@ -1534,7 +1534,7 @@ pub async fn sync_wallet(client: BlockchainClient, wallet: &mut Wallet) -> Resul
             parallel_requests,
         } => {
             let update = client
-                .sync(request, parallel_requests)
+                .sync(request, *parallel_requests)
                 .await
                 .map_err(|e| *e)?;
             wallet
@@ -1549,7 +1549,7 @@ pub async fn sync_wallet(client: BlockchainClient, wallet: &mut Wallet) -> Resul
             // reload the last 200 blocks in case of a reorg
             let emitter_height = wallet_cp.height().saturating_sub(200);
             let mut emitter = Emitter::new(
-                &*client,
+                client.as_ref(),
                 wallet_cp,
                 emitter_height,
                 wallet
@@ -1600,7 +1600,7 @@ pub async fn sync_wallet(client: BlockchainClient, wallet: &mut Wallet) -> Resul
 ))]
 /// Broadcasts a given transaction using the blockchain client.
 pub async fn broadcast_transaction(
-    client: BlockchainClient,
+    client: &BlockchainClient,
     tx: Transaction,
 ) -> Result<Txid, Error> {
     match client {
@@ -1627,38 +1627,15 @@ pub async fn broadcast_transaction(
 
         #[cfg(feature = "cbf")]
         KyotoClient { client } => {
-            let LightClient {
-                requester,
-                mut info_subscriber,
-                mut warning_subscriber,
-                update_subscriber: _,
-                node,
-            } = *client;
-
-            let subscriber = tracing_subscriber::FmtSubscriber::new();
-            tracing::subscriber::set_global_default(subscriber)
-                .map_err(|e| Error::Generic(format!("SetGlobalDefault error: {e}")))?;
-
-            tokio::task::spawn(async move { node.run().await });
-            tokio::task::spawn(async move {
-                select! {
-                    info = info_subscriber.recv() => {
-                        if let Some(info) = info {
-                            tracing::info!("{info}");
-                        }
-                    },
-                    warn = warning_subscriber.recv() => {
-                        if let Some(warn) = warn {
-                            tracing::warn!("{warn}");
-                        }
-                    }
-                }
-            });
             let txid = tx.compute_txid();
-            let wtxid = requester.broadcast_random(tx.clone()).await.map_err(|_| {
-                tracing::warn!("Broadcast was unsuccessful");
-                Error::Generic("Transaction broadcast timed out after 30 seconds".into())
-            })?;
+            let wtxid = client
+                .requester
+                .broadcast_random(tx.clone())
+                .await
+                .map_err(|_| {
+                    tracing::warn!("Broadcast was unsuccessful");
+                    Error::Generic("Transaction broadcast timed out after 30 seconds".into())
+                })?;
             tracing::info!("Successfully broadcast WTXID: {wtxid}");
             Ok(txid)
         }