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

Author: tvpeter

src/handlers.rs

@@ -34,7 +34,7 @@ use bdk_wallet::{KeychainKind, SignOptions, Wallet};
 #[cfg(feature = "cbf")]
 use {
     crate::utils::BlockchainClient::KyotoClient,
-    bdk_kyoto::{LightClient, RequesterExt, TxBroadcastPolicy::RandomPeer},
+    bdk_kyoto::{LightClient, RequesterExt, TxBroadcast, TxBroadcastPolicy::RandomPeer},
 };
 
 use bdk_wallet::keys::DescriptorKey::Secret;
@@ -688,7 +688,7 @@ pub(crate) async fn handle_online_wallet_subcommand(
                         requester,
                         log_subscriber,
                         warning_subscriber,
-                        update_subscriber: _,
+                        mut update_subscriber,
                         node,
                     } = client;
 
@@ -701,15 +701,90 @@ pub(crate) async fn handle_online_wallet_subcommand(
                         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(bdk_kyoto::TxBroadcast {
+                        .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()
                 }

src/utils.rs

@@ -194,9 +194,15 @@ pub(crate) fn new_blockchain_client(
 
         #[cfg(feature = "cbf")]
         ClientType::Cbf => {
+            // default to skipping 10 blocks away from the wallet chain tip
+            let skip_blocks = if wallet_opts.compactfilter_opts.skip_blocks == 0 {
+                wallet.local_chain().tip().height() - 10
+            } else {
+                wallet_opts.compactfilter_opts.skip_blocks
+            };
             let client = LightClientBuilder::new()
                 .connections(wallet_opts.compactfilter_opts.conn_count)
-                .scan_after(wallet_opts.compactfilter_opts.skip_blocks)
+                .scan_after(skip_blocks)
                 .build(wallet)?;
             BlockchainClient::KyotoClient { client }
         }