feat(cbf): update bdk-kyoto to 0.9.0

tvpeter · tvpeter · commit 6755483b89c5 · 2025-05-07T15:01:24.000+01:00
- refactor syncing into a fn
- made `skip-blocks` optional and removed default
value to use bdk-kyoto Sync scan type
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -25,7 +25,7 @@ tokio = { version = "1", features = ["full"] }
 bdk_bitcoind_rpc = { version = "0.18.0", optional = true }
 bdk_electrum = { version = "0.21.0", optional = true }
 bdk_esplora = { version = "0.20.1", features = ["async-https", "tokio"], optional = true }
-bdk_kyoto = { version = "0.7.1", optional = true }
+bdk_kyoto = { version = "0.9.0", optional = true }
 shlex = {  version = "1.3.0", optional = true }
 tracing = "0.1.41"
 tracing-subscriber = "0.3.19"
diff --git a/src/commands.rs b/src/commands.rs
@@ -20,12 +20,7 @@ use bdk_wallet::bitcoin::{
 };
 use clap::{value_parser, Args, Parser, Subcommand, ValueEnum};
 
-#[cfg(any(
-    feature = "cbf",
-    feature = "electrum",
-    feature = "esplora",
-    feature = "rpc"
-))]
+#[cfg(any(feature = "electrum", feature = "esplora", feature = "rpc"))]
 use crate::utils::parse_proxy_auth;
 use crate::utils::{parse_address, parse_outpoint, parse_recipient};
 
@@ -214,7 +209,7 @@ pub struct WalletOpts {
 }
 
 /// Options to configure a SOCKS5 proxy for a blockchain client connection.
-#[cfg(any(feature = "cbf", feature = "electrum", feature = "esplora"))]
+#[cfg(any(feature = "electrum", feature = "esplora"))]
 #[derive(Debug, Args, Clone, PartialEq, Eq)]
 pub struct ProxyOpts {
     /// Sets the SOCKS5 proxy for a blockchain client.
@@ -244,17 +239,12 @@ pub struct ProxyOpts {
 #[derive(Debug, Args, Clone, PartialEq, Eq)]
 pub struct CompactFilterOpts {
     /// Sets the number of parallel node connections.
-    #[clap(name = "CONNECTIONS", long = "cbf-conn-count", default_value = "4")]
+    #[clap(name = "CONNECTIONS", long = "cbf-conn-count", default_value = "4", value_parser = value_parser!(u8).range(1..=15))]
     pub conn_count: u8,
 
     /// Optionally skip initial `skip_blocks` blocks.
-    #[clap(
-        env = "SKIP_BLOCKS",
-        short = 'k',
-        long = "cbf-skip-blocks",
-        default_value = "0"
-    )]
-    pub skip_blocks: u32,
+    #[clap(env = "SKIP_BLOCKS", short = 'k', long = "cbf-skip-blocks")]
+    pub skip_blocks: Option<u32>,
 }
 
 /// Wallet subcommands that can be issued without a blockchain backend.
diff --git a/src/error.rs b/src/error.rs
@@ -87,6 +87,6 @@ pub enum BDKCliError {
     BitcoinCoreRpcError(#[from] bdk_bitcoind_rpc::bitcoincore_rpc::Error),
 
     #[cfg(feature = "cbf")]
-    #[error("Kyoto-cbf error: {0}")]
-    SqlInitializationError(#[from] bdk_kyoto::kyoto::db::error::SqlInitializationError),
+    #[error("BDK-Kyoto error: {0}")]
+    BuilderError(#[from] bdk_kyoto::builder::BuilderError),
 }
diff --git a/src/handlers.rs b/src/handlers.rs
@@ -420,102 +420,7 @@ pub(crate) async fn handle_online_wallet_subcommand(
                 }
                 #[cfg(feature = "cbf")]
                 KyotoClient { client } => {
-                    let LightClient {
-                        requester,
-                        log_subscriber,
-                        warning_subscriber,
-                        mut 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 {
-                        trace_logger(log_subscriber, warning_subscriber).await
-                    });
-
-                    if !requester.is_running().await {
-                        tracing::error!("Kyoto node is not running");
-                        return Err(Error::Generic("Kyoto node failed to start".to_string()));
-                    }
-                    tracing::info!("Kyoto node is running");
-
-                    tracing::info!(
-                        "Initial derivation indices: External={}, Internal={}",
-                        wallet.derivation_index(KeychainKind::External).unwrap_or(0),
-                        wallet.derivation_index(KeychainKind::Internal).unwrap_or(0)
-                    );
-
-                    tracing::info!("Adding wallet scripts to node");
-                    requester.add_revealed_scripts(&wallet).await.map_err(|e| {
-                        Error::Generic(format!("Failed to add wallet scripts: {}", e))
-                    })?;
-
-                    let mut updates_applied = false;
-                    let mut initial_tx_count = wallet.transactions().count();
-                    let mut iteration_count = 0;
-                    const MAX_ITERATIONS: u32 = 600;
-
-                    loop {
-                        iteration_count += 1;
-                        tracing::debug!(
-                            "Loop iteration {}, updates_applied: {}",
-                            iteration_count,
-                            updates_applied
-                        );
-
-                        if iteration_count > MAX_ITERATIONS && !updates_applied {
-                            tracing::error!(
-                                "Timeout: No updates received after {} iterations",
-                                MAX_ITERATIONS
-                            );
-                            return Err(Error::Generic(
-                                "Sync timed out: no updates received".to_string(),
-                            ));
-                        }
-
-                        match update_subscriber.update().await {
-                            Some(update) => {
-                                tracing::info!("Received update: applying to wallet");
-                                wallet.apply_update(update).map_err(|e| {
-                                    Error::Generic(format!("Failed to apply update: {}", e))
-                                })?;
-                                updates_applied = true;
-
-                                tracing::info!(
-                                    "Tx count: {}, Balance: {}, Chain tip: {}",
-                                    wallet.transactions().count(),
-                                    wallet.balance().total().to_sat(),
-                                    wallet.local_chain().tip().height()
-                                );
-
-                                let current_tx_count = wallet.transactions().count();
-                                if current_tx_count == initial_tx_count && updates_applied {
-                                    tracing::info!("No new transactions found; sync complete");
-                                    break;
-                                }
-                                initial_tx_count = current_tx_count;
-                            }
-                            None => {
-                                tracing::debug!("No update received, waiting");
-                                if updates_applied {
-                                    tracing::info!("No further updates available");
-                                    break;
-                                }
-                                // Wait briefly to avoid tight loop
-                                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-                            }
-                        }
-                    }
-
-                    tracing::info!(
-                        "Sync completed: tx_count={}, balance={}",
-                        wallet.transactions().count(),
-                        wallet.balance().total().to_sat()
-                    );
+                    sync_kyoto_client(wallet, client).await?;
                 }
             }
             Ok(json!({}))
@@ -583,61 +488,7 @@ pub(crate) async fn handle_online_wallet_subcommand(
                 }
                 #[cfg(feature = "cbf")]
                 KyotoClient { client } => {
-                    let LightClient {
-                        requester,
-                        log_subscriber,
-                        warning_subscriber,
-                        mut 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 {
-                        trace_logger(log_subscriber, warning_subscriber).await
-                    });
-
-                    requester
-                        .add_revealed_scripts(&wallet)
-                        .await
-                        .map_err(|e| Error::Generic(format!("Failed to add script: {}", e)))?;
-
-                    let mut updates_applied = false;
-                    let mut initial_tx_count = wallet.transactions().count();
-
-                    loop {
-                        match update_subscriber.update().await {
-                            Some(update) => {
-                                wallet.apply_update(update).map_err(|e| {
-                                    Error::Generic(format!("Failed to apply update: {}", e))
-                                })?;
-                                updates_applied = true;
-
-                                tracing::info!(
-                                    "Chain tip: {}, Transactions: {}",
-                                    wallet.local_chain().tip().height(),
-                                    wallet.transactions().count()
-                                );
-
-                                let current_tx_count = wallet.transactions().count();
-                                if current_tx_count == initial_tx_count && updates_applied {
-                                    tracing::info!("No new transactions found; sync complete");
-                                    break;
-                                }
-                                initial_tx_count = current_tx_count;
-                            }
-                            None => {
-                                if updates_applied {
-                                    tracing::info!("No further updates available");
-                                    break;
-                                }
-                                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-                            }
-                        }
-                    }
+                    sync_kyoto_client(wallet, client).await?;
                 }
             }
             Ok(json!({}))
@@ -684,109 +535,7 @@ pub(crate) async fn handle_online_wallet_subcommand(
 
                 #[cfg(feature = "cbf")]
                 KyotoClient { client } => {
-                    let LightClient {
-                        requester,
-                        log_subscriber,
-                        warning_subscriber,
-                        mut 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 {
-                        trace_logger(log_subscriber, warning_subscriber).await
-                    });
-
-                    // Wait for peer connections to establish
-                    const PEER_CONNECTION_DELAY: u64 = 60;
-                    tracing::info!(
-                        "Waiting {} seconds for peer connections",
-                        PEER_CONNECTION_DELAY
-                    );
-                    tokio::time::sleep(std::time::Duration::from_secs(PEER_CONNECTION_DELAY)).await;
-
-                    tracing::info!("Proceeding with transaction broadcast");
-
-                    // Broadcast the transaction
-                    let broadcasted_txid = tx.compute_txid();
-                    tracing::info!("Broadcasting transaction: {}", broadcasted_txid);
-                    requester
-                        .broadcast_tx(TxBroadcast {
-                            tx: tx.clone(),
-                            broadcast_policy: RandomPeer,
-                        })
-                        .await
-                        .map_err(|e| {
-                            tracing::error!("Failed to broadcast transaction: {}", e);
-                            Error::Generic(format!("Failed to broadcast transaction: {}", e))
-                        })?;
-                    tracing::info!("Transaction broadcasted successfully");
-
-                    // Perform a sync to ensure transaction propagation
-                    tracing::info!("Starting sync to confirm transaction in mempool",);
-                    requester.add_revealed_scripts(&wallet).await.map_err(|e| {
-                        Error::Generic(format!("Failed to add wallet scripts: {}", e))
-                    })?;
-
-                    let chain_tip = wallet.local_chain().tip();
-                    tracing::info!(
-                        "Starting sync to confirm transaction {} in mempool from chain tip: height={}, hash={}",
-                        broadcasted_txid,
-                        chain_tip.height(),
-                        chain_tip.hash()
-                    );
-
-                    requester.add_revealed_scripts(&wallet).await.map_err(|e| {
-                        Error::Generic(format!("Failed to add wallet scripts: {}", e))
-                    })?;
-
-                    let mut updates_applied = false;
-                    loop {
-                        match update_subscriber.update().await {
-                            Some(update) => {
-                                wallet.apply_update(update).map_err(|e| {
-                                    Error::Generic(format!("Failed to apply update: {}", e))
-                                })?;
-                                updates_applied = true;
-                                tracing::info!(
-                                    "Applied update: tx_count={}, balance={}, chain_tip={}",
-                                    wallet.transactions().count(),
-                                    wallet.balance().total().to_sat(),
-                                    wallet.local_chain().tip().height()
-                                );
-                                break;
-                            }
-                            None => {
-                                if updates_applied {
-                                    tracing::info!("No further updates available during sync");
-                                    break;
-                                }
-                                tracing::debug!("No update received, waiting");
-                                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-                            }
-                        }
-                    }
-
-                    if !wallet
-                        .transactions()
-                        .any(|tx| tx.tx_node.txid == broadcasted_txid)
-                    {
-                        tracing::warn!(
-                            "Transaction {} not found in wallet after sync; may not be propagated",
-                            broadcasted_txid
-                        );
-                    } else {
-                        tracing::info!(
-                            "Sync completed; transaction {} likely propagated",
-                            broadcasted_txid
-                        );
-                    }
-
-                    tx.compute_txid()
+                    unimplemented!()
                 }
             };
             Ok(json!({ "txid": txid }))
diff --git a/src/utils.rs b/src/utils.rs

Original file line number	Diff line number	Diff line change
`@@ -87,6 +87,6 @@ pub enum BDKCliError {`
`87`	`87`	`BitcoinCoreRpcError(#[from] bdk_bitcoind_rpc::bitcoincore_rpc::Error),`
`88`	`88`
`89`	`89`	`#[cfg(feature = "cbf")]`
`90`		`- #[error("Kyoto-cbf error: {0}")]`
`91`		`- SqlInitializationError(#[from] bdk_kyoto::kyoto::db::error::SqlInitializationError),`
	`90`	`+ #[error("BDK-Kyoto error: {0}")]`
	`91`	`+ BuilderError(#[from] bdk_kyoto::builder::BuilderError),`
`92`	`92`	`}`