feat(cbf): add cbf feature using bdk_kyoto

- enable full_scan and sync operations

[issue: #172]

feat(cbf): update broadcasting tx

- add wait time for node to connect to peers
before broadcasting tx
- add sync chain starting from 10 blocks below
the wallet tip to ensure tx is propagated
- update code_coverage workflow to cover cbf
feature

feat(cbf): update bdk-kyoto to 0.9.0

- refactor syncing into a fn
- made `skip-blocks` optional and removed default
value to use bdk-kyoto Sync scan type

feat(cbf): remove looping for kyoto sync

- remove looping for kyoto client sync operations
- fix compiler warnings

diff --git a/.github/workflows/code_coverage.yml b/.github/workflows/code_coverage.yml
index 9dd9cd1cb09bf3d869f0293172ed4e39a76b56c1..77b34bb29a71d7ed1f31811fd3d0da848e4d9a73 100644 (file)
--- a/.github/workflows/code_coverage.yml
+++ b/.github/workflows/code_coverage.yml
@@ -35,10 +35,9 @@ jobs:
       - name: Test Esplora
         run: cargo test --features esplora
 
-      # Temporarily disable compact filters
-      #- name: Test Compact Filters
-      #  run: cargo test --features compact_filters
-
+      - name: Test Cbf
+        run: cargo test --features cbf
+        
       - name: Test RPC
         run: cargo test --features rpc
 

diff --git a/Cargo.lock b/Cargo.lock
index 0776f87212640fe01690c84e5e24161b240264d1..35d240ac80e979eb36690dff49596f2063272363 100644 (file)
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,6 +1,6 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
-version = 3
+version = 4
 
 [[package]]
 name = "addr2line"
@@ -194,6 +194,8 @@ dependencies = [
  "shlex",
  "thiserror 2.0.12",
  "tokio",
+ "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]
@@ -256,9 +258,9 @@ dependencies = [
 
 [[package]]
 name = "bdk_kyoto"
-version = "0.7.1"
+version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "669e8d613e93c400ae82596404732fbf521cee41a3c8b96a6c011a4dea21a5dc"
+checksum = "510abdf0efa06d5bc83a48af90ca43718ea8adf1cf660c663ca313ca0846144a"
 dependencies = [
  "bdk_wallet",
  "kyoto-cbf",
@@ -310,9 +312,9 @@ dependencies = [
 
 [[package]]
 name = "bip324"
-version = "0.6.0"
+version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b443a76f86143c093b211628be683ee592a097d316db6b90f723ed816bde1a49"
+checksum = "53157fcb2d6ec2851c7602d0690536d0b79209e393972cb2b36bd5d72dbd1879"
 dependencies = [
  "bitcoin",
  "bitcoin_hashes 0.15.0",
@@ -1178,9 +1180,9 @@ dependencies = [
 
 [[package]]
 name = "kyoto-cbf"
-version = "0.8.0"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6943c874dd9f43175b3751d091d11f43a0d4c9a9bc10751c0f19a70c1862d64e"
+checksum = "a71eba746c4c4936a1b75336560b40ebe1145aa5b87cc90bc0ccfeacf4c49e79"
 dependencies = [
  "bip324",
  "bitcoin",
@@ -1213,7 +1215,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fc2f4eb4bc735547cfed7c0a4922cbd04a4655978c09b54f1f7b228750664c34"
 dependencies = [
  "cfg-if",
- "windows-targets 0.48.5",
+ "windows-targets 0.52.6",
 ]
 
 [[package]]
@@ -1357,6 +1359,16 @@ dependencies = [
  "minimal-lexical",
 ]
 
+[[package]]
+name = "nu-ansi-term"
+version = "0.46.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84"
+dependencies = [
+ "overload",
+ "winapi",
+]
+
 [[package]]
 name = "num_cpus"
 version = "1.16.0"
@@ -1432,6 +1444,12 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
 
+[[package]]
+name = "overload"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
+
 [[package]]
 name = "parking_lot"
 version = "0.12.3"
@@ -1906,6 +1924,15 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "sharded-slab"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6"
+dependencies = [
+ "lazy_static",
+]
+
 [[package]]
 name = "shlex"
 version = "1.3.0"
@@ -2050,6 +2077,16 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "thread_local"
+version = "1.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b9ef9bad013ada3808854ceac7b46812a6465ba368859a37e2100283d2d719c"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+]
+
 [[package]]
 name = "tinyvec"
 version = "1.9.0"
@@ -2143,9 +2180,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0"
 dependencies = [
  "pin-project-lite",
+ "tracing-attributes",
  "tracing-core",
 ]
 
+[[package]]
+name = "tracing-attributes"
+version = "0.1.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "395ae124c09f9e6918a2310af6038fba074bcf474ac352496d5910dd59a2226d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "tracing-core"
 version = "0.1.33"
@@ -2153,6 +2202,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e672c95779cf947c5311f83787af4fa8fffd12fb27e4993211a84bdfd9610f9c"
 dependencies = [
  "once_cell",
+ "valuable",
+]
+
+[[package]]
+name = "tracing-log"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3"
+dependencies = [
+ "log",
+ "once_cell",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-subscriber"
+version = "0.3.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8189decb5ac0fa7bc8b96b7cb9b2701d60d48805aca84a238004d665fcc4008"
+dependencies = [
+ "nu-ansi-term",
+ "sharded-slab",
+ "smallvec",
+ "thread_local",
+ "tracing-core",
+ "tracing-log",
 ]
 
 [[package]]
@@ -2217,6 +2292,12 @@ version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
 
+[[package]]
+name = "valuable"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
+
 [[package]]
 name = "vcpkg"
 version = "0.2.15"

diff --git a/Cargo.toml b/Cargo.toml
index e1fe7425df6b4977a0d5ee58b79f9938ed3ebd65..f3fd34afc0b94b4484536c1b7dfd5c588d5f98d0 100644 (file)
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -25,8 +25,10 @@ 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"
 
 [features]
 default = ["repl", "sqlite"]

diff --git a/src/commands.rs b/src/commands.rs
index bac916c33f427c3b0b8e5a6d6ed12412db36e23f..67f845374e93cac6904c221d7f2380a2eddaa9ba 100644 (file)
--- 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};
 
@@ -133,7 +128,12 @@ pub enum DatabaseType {
     Sqlite,
 }
 
-#[cfg(any(feature = "electrum", feature = "esplora", feature = "rpc"))]
+#[cfg(any(
+    feature = "electrum",
+    feature = "esplora",
+    feature = "rpc",
+    feature = "cbf"
+))]
 #[derive(Clone, ValueEnum, Debug, Eq, PartialEq)]
 pub enum ClientType {
     #[cfg(feature = "electrum")]
@@ -142,6 +142,8 @@ pub enum ClientType {
     Esplora,
     #[cfg(feature = "rpc")]
     Rpc,
+    #[cfg(feature = "cbf")]
+    Cbf,
 }
 
 /// Config options wallet operations can take.
@@ -159,7 +161,12 @@ pub struct WalletOpts {
     /// Sets the descriptor to use for internal/change addresses.
     #[arg(env = "INT_DESCRIPTOR", short = 'i', long)]
     pub int_descriptor: Option<String>,
-    #[cfg(any(feature = "electrum", feature = "esplora", feature = "rpc"))]
+    #[cfg(any(
+        feature = "electrum",
+        feature = "esplora",
+        feature = "rpc",
+        feature = "cbf"
+    ))]
     #[arg(env = "CLIENT_TYPE", short = 'c', long, value_enum, required = true)]
     pub client_type: ClientType,
     #[cfg(feature = "sqlite")]
@@ -196,10 +203,13 @@ pub struct WalletOpts {
     /// Sets an optional cookie authentication.
     #[arg(env = "COOKIE")]
     pub cookie: Option<String>,
+    #[cfg(feature = "cbf")]
+    #[clap(flatten)]
+    pub compactfilter_opts: CompactFilterOpts,
 }
 
 /// 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.
@@ -228,26 +238,13 @@ pub struct ProxyOpts {
 #[cfg(feature = "cbf")]
 #[derive(Debug, Args, Clone, PartialEq, Eq)]
 pub struct CompactFilterOpts {
-    /// Sets the full node network address.
-    #[clap(
-        env = "ADDRESS:PORT",
-        long = "cbf-node",
-        default_value = "127.0.0.1:18444"
-    )]
-    pub address: Vec<String>,
-
     /// Sets the number of parallel node connections.
-    #[clap(name = "CONNECTIONS", long = "cbf-conn-count", default_value = "4")]
-    pub conn_count: usize,
+    #[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: usize,
+    #[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
index ebacc8de8da609082f51547ba193db809d902366..27606d730f75cd1c6f07ab7126365ca2f1680823 100644 (file)
--- a/src/error.rs
+++ b/src/error.rs
@@ -85,4 +85,8 @@ pub enum BDKCliError {
     #[cfg(feature = "rpc")]
     #[error("RPC error: {0}")]
     BitcoinCoreRpcError(#[from] bdk_bitcoind_rpc::bitcoincore_rpc::Error),
+
+    #[cfg(feature = "cbf")]
+    #[error("BDK-Kyoto error: {0}")]
+    BuilderError(#[from] bdk_kyoto::builder::BuilderError),
 }

diff --git a/src/handlers.rs b/src/handlers.rs
index 0873a23b9e5e8673e34568a930866070815cf5b0..1b7bf25c5dee18b4d797f326350de6f51417ddad 100644 (file)
--- a/src/handlers.rs
+++ b/src/handlers.rs
@@ -11,28 +11,16 @@
 //! This module describes all the command handling logic used by bdk-cli.
 
 use crate::commands::OfflineWalletSubCommand::*;
-#[cfg(any(
-    feature = "electrum",
-    feature = "esplora",
-    feature = "cbf",
-    feature = "rpc"
-))]
-use crate::commands::OnlineWalletSubCommand::*;
 use crate::commands::*;
 use crate::error::BDKCliError as Error;
+#[cfg(feature = "cbf")]
+use crate::utils::BlockchainClient::KyotoClient;
 use crate::utils::*;
 use bdk_wallet::bip39::{Language, Mnemonic};
 use bdk_wallet::bitcoin::bip32::{DerivationPath, KeySource};
 use bdk_wallet::bitcoin::consensus::encode::serialize_hex;
 use bdk_wallet::bitcoin::script::PushBytesBuf;
 use bdk_wallet::bitcoin::Network;
-#[cfg(any(
-    feature = "electrum",
-    feature = "esplora",
-    feature = "cbf",
-    feature = "rpc"
-))]
-use bdk_wallet::bitcoin::Transaction;
 use bdk_wallet::bitcoin::{secp256k1::Secp256k1, Txid};
 use bdk_wallet::bitcoin::{Amount, FeeRate, Psbt, Sequence};
 use bdk_wallet::descriptor::Segwitv0;
@@ -51,7 +39,7 @@ use bdk_wallet::keys::{DerivableKey, DescriptorKey, ExtendedKey, GeneratableKey,
 use bdk_wallet::miniscript::miniscript;
 use serde_json::json;
 use std::collections::BTreeMap;
-#[cfg(any(feature = "electrum", feature = "esplora", feature = "cbf",))]
+#[cfg(any(feature = "electrum", feature = "esplora"))]
 use std::collections::HashSet;
 use std::convert::TryFrom;
 #[cfg(feature = "repl")]
@@ -67,14 +55,10 @@ use bdk_wallet::bitcoin::base64::prelude::*;
     feature = "cbf",
     feature = "rpc"
 ))]
-use bdk_wallet::bitcoin::consensus::Decodable;
-#[cfg(any(
-    feature = "electrum",
-    feature = "esplora",
-    feature = "cbf",
-    feature = "rpc"
-))]
-use bdk_wallet::bitcoin::hex::FromHex;
+use {
+    crate::commands::OnlineWalletSubCommand::*,
+    bdk_wallet::bitcoin::{consensus::Decodable, hex::FromHex, Transaction},
+};
 #[cfg(feature = "esplora")]
 use {crate::utils::BlockchainClient::Esplora, bdk_esplora::EsploraAsyncExt};
 #[cfg(feature = "rpc")]
@@ -431,6 +415,10 @@ pub(crate) async fn handle_online_wallet_subcommand(
                     let mempool_txs = emitter.mempool()?;
                     wallet.apply_unconfirmed_txs(mempool_txs);
                 }
+                #[cfg(feature = "cbf")]
+                KyotoClient { client } => {
+                    sync_kyoto_client(wallet, client).await?;
+                }
             }
             Ok(json!({}))
         }
@@ -495,6 +483,10 @@ pub(crate) async fn handle_online_wallet_subcommand(
                     let mempool_txs = emitter.mempool()?;
                     wallet.apply_unconfirmed_txs(mempool_txs);
                 }
+                #[cfg(feature = "cbf")]
+                KyotoClient { client } => {
+                    sync_kyoto_client(wallet, client).await?;
+                }
             }
             Ok(json!({}))
         }
@@ -537,6 +529,11 @@ pub(crate) async fn handle_online_wallet_subcommand(
                 RpcClient { client } => client
                     .send_raw_transaction(&tx)
                     .map_err(|e| Error::Generic(e.to_string()))?,
+
+                #[cfg(feature = "cbf")]
+                KyotoClient { client: _ } => {
+                    unimplemented!()
+                }
             };
             Ok(json!({ "txid": txid }))
         }
@@ -683,7 +680,6 @@ pub(crate) async fn handle_command(cli_opts: CliOpts) -> Result<String, Error> {
             wallet_opts,
             subcommand: WalletSubCommand::OnlineWalletSubCommand(online_subcommand),
         } => {
-            let blockchain_client = new_blockchain_client(&wallet_opts)?;
             let network = cli_opts.network;
             #[cfg(feature = "sqlite")]
             let result = {
@@ -701,6 +697,8 @@ pub(crate) async fn handle_command(cli_opts: CliOpts) -> Result<String, Error> {
                 };
 
                 let mut wallet = new_persisted_wallet(network, &mut persister, &wallet_opts)?;
+                let blockchain_client = new_blockchain_client(&wallet_opts, &wallet)?;
+
                 let result = handle_online_wallet_subcommand(
                     &mut wallet,
                     blockchain_client,
@@ -847,7 +845,8 @@ async fn respond(
         ReplSubCommand::Wallet {
             subcommand: WalletSubCommand::OnlineWalletSubCommand(online_subcommand),
         } => {
-            let blockchain = new_blockchain_client(wallet_opts).map_err(|e| e.to_string())?;
+            let blockchain =
+                new_blockchain_client(wallet_opts, &wallet).map_err(|e| e.to_string())?;
             let value = handle_online_wallet_subcommand(wallet, blockchain, online_subcommand)
                 .await
                 .map_err(|e| e.to_string())?;

diff --git a/src/utils.rs b/src/utils.rs
index d2aff1a691d795daadefdbb77319f3ff90ff39c2..feb08b481d1d70404f0069c089fd58e8a2c09eb0 100644 (file)
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -16,9 +16,21 @@ use std::str::FromStr;
 use std::path::{Path, PathBuf};
 
 use crate::commands::WalletOpts;
+#[cfg(feature = "cbf")]
+use bdk_kyoto::{
+    builder::NodeBuilder,
+    Info, LightClient, NodeBuilderExt, Receiver,
+    ScanType::{Recovery, Sync},
+    UnboundedReceiver, Warning,
+};
 use bdk_wallet::bitcoin::{Address, Network, OutPoint, ScriptBuf};
 
-#[cfg(any(feature = "electrum", feature = "esplora", feature = "rpc"))]
+#[cfg(any(
+    feature = "electrum",
+    feature = "esplora",
+    feature = "rpc",
+    feature = "cbf"
+))]
 use crate::commands::ClientType;
 
 use bdk_wallet::Wallet;
@@ -39,12 +51,7 @@ pub(crate) fn parse_recipient(s: &str) -> Result<(ScriptBuf, u64), String> {
     Ok((addr.script_pubkey(), val))
 }
 
-#[cfg(any(
-    feature = "electrum",
-    feature = "cbf",
-    feature = "esplora",
-    feature = "rpc"
-))]
+#[cfg(any(feature = "electrum", feature = "esplora", feature = "rpc"))]
 /// Parse the proxy (Socket:Port) argument from the cli input.
 pub(crate) fn parse_proxy_auth(s: &str) -> Result<(String, String), Error> {
     let parts: Vec<_> = s.split(':').collect();
@@ -132,7 +139,9 @@ pub(crate) enum BlockchainClient {
     RpcClient {
         client: Box<bdk_bitcoind_rpc::bitcoincore_rpc::Client>,
     },
-    // TODO cbf
+
+    #[cfg(feature = "cbf")]
+    KyotoClient { client: LightClient },
 }
 
 #[cfg(any(
@@ -142,7 +151,11 @@ pub(crate) enum BlockchainClient {
     feature = "cbf",
 ))]
 /// Create a new blockchain from the wallet configuration options.
-pub(crate) fn new_blockchain_client(wallet_opts: &WalletOpts) -> Result<BlockchainClient, Error> {
+pub(crate) fn new_blockchain_client(
+    wallet_opts: &WalletOpts,
+    wallet: &Wallet,
+) -> Result<BlockchainClient, Error> {
+    #[cfg(any(feature = "electrum", feature = "esplora", feature = "rpc"))]
     let url = wallet_opts.url.as_str();
     let client = match wallet_opts.client_type {
         #[cfg(feature = "electrum")]
@@ -178,6 +191,20 @@ pub(crate) fn new_blockchain_client(wallet_opts: &WalletOpts) -> Result<Blockcha
                 client: Box::new(client),
             }
         }
+
+        #[cfg(feature = "cbf")]
+        ClientType::Cbf => {
+            let scan_type = match wallet_opts.compactfilter_opts.skip_blocks {
+                Some(from_height) => Recovery { from_height },
+                None => Sync,
+            };
+
+            let client = NodeBuilder::new(wallet.network())
+                .required_peers(wallet_opts.compactfilter_opts.conn_count)
+                .build_with_wallet(wallet, scan_type)?;
+
+            BlockchainClient::KyotoClient { client }
+        }
     };
     Ok(client)
 }
@@ -263,3 +290,79 @@ pub(crate) fn new_wallet(network: Network, wallet_opts: &WalletOpts) -> Result<W
         )),
     }
 }
+
+#[cfg(feature = "cbf")]
+pub async fn trace_logger(
+    mut log_subscriber: Receiver<String>,
+    mut info_subcriber: Receiver<Info>,
+    mut warning_subscriber: UnboundedReceiver<Warning>,
+) {
+    loop {
+        tokio::select! {
+            log = log_subscriber.recv() => {
+                if let Some(log) = log {
+                    tracing::info!("{log}")
+                }
+            }
+            info = info_subcriber.recv() => {
+                if let Some(info) = info {
+                    tracing::info!("{info}")
+                }
+            }
+            warn = warning_subscriber.recv() => {
+                if let Some(warn) = warn {
+                    tracing::warn!("{warn}")
+                }
+            }
+        }
+    }
+}
+
+// Handle Kyoto Client sync
+#[cfg(feature = "cbf")]
+pub async fn sync_kyoto_client(wallet: &mut Wallet, client: LightClient) -> Result<(), Error> {
+    let LightClient {
+        requester,
+        log_subscriber,
+        info_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, info_subscriber, warning_subscriber).await
+    });
+
+    if !requester.is_running() {
+        tracing::error!("Kyoto node is not running");
+        return Err(Error::Generic("Kyoto node failed to start".to_string()));
+    }
+    tracing::info!("Kyoto node is running");
+
+    let update = update_subscriber.update().await;
+    tracing::info!("Received update: applying to wallet");
+    wallet
+        .apply_update(update)
+        .map_err(|e| Error::Generic(format!("Failed to apply update: {}", e)))?;
+
+    tracing::info!(
+        "Chain tip: {}, Transactions: {}, Balance: {}",
+        wallet.local_chain().tip().height(),
+        wallet.transactions().count(),
+        wallet.balance().total().to_sat()
+    );
+
+    tracing::info!(
+        "Sync completed: tx_count={}, balance={}",
+        wallet.transactions().count(),
+        wallet.balance().total().to_sat()
+    );
+
+    Ok(())
+}