Multiple wallet syncing

examples/multi.rs

@@ -0,0 +1,107 @@
+use bdk_kyoto::builder::{Builder, BuilderExt};
+use bdk_kyoto::{Info, Receiver, ScanType, UnboundedReceiver, Warning};
+use bdk_wallet::bitcoin::Network;
+use bdk_wallet::chain::DescriptorExt;
+use bdk_wallet::{KeychainKind, Wallet};
+use tokio::select;
+
+const RECV_ONE: &str = "wpkh([9122d9e0/84'/1'/0']tpubDCYVtmaSaDzTxcgvoP5AHZNbZKZzrvoNH9KARep88vESc6MxRqAp4LmePc2eeGX6XUxBcdhAmkthWTDqygPz2wLAyHWisD299Lkdrj5egY6/0/*)";
+const CHANGE_ONE: &str = "wpkh([9122d9e0/84'/1'/0']tpubDCYVtmaSaDzTxcgvoP5AHZNbZKZzrvoNH9KARep88vESc6MxRqAp4LmePc2eeGX6XUxBcdhAmkthWTDqygPz2wLAyHWisD299Lkdrj5egY6/1/*)";
+const RECV_TWO: &str = "tr([7d94197e/86'/1'/0']tpubDCyQVJj8KzjiQsFjmb3KwECVXPvMwvAxxZGCP9XmWSopmjW3bCV3wD7TgxrUhiGSueDS1MU5X1Vb1YjYcp8jitXc5fXfdC1z68hDDEyKRNr/0/*)";
+const CHANGE_TWO: &str = "tr([7d94197e/86'/1'/0']tpubDCyQVJj8KzjiQsFjmb3KwECVXPvMwvAxxZGCP9XmWSopmjW3bCV3wD7TgxrUhiGSueDS1MU5X1Vb1YjYcp8jitXc5fXfdC1z68hDDEyKRNr/1/*)";
+const NETWORK: Network = Network::Signet;
+
+/* Sync multiple BDK wallets */
+
+async fn traces(
+    mut info_subscriber: Receiver<Info>,
+    mut warning_subscriber: UnboundedReceiver<Warning>,
+) {
+    loop {
+        select! {
+            info = info_subscriber.recv() => {
+                if let Some(info) = info {
+                    match info {
+                        Info::Progress(p) => {
+                            tracing::info!("chain height: {}, filter download progress: {}%", p.chain_height(), p.percentage_complete());
+                        },
+                        Info::BlockReceived(b) => {
+                            tracing::info!("downloaded block: {b}");
+                        },
+                        _ => (),
+                    }
+                }
+            }
+            warn = warning_subscriber.recv() => {
+                if let Some(warn) = warn {
+                    tracing::warn!("{warn}")
+                }
+            }
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let subscriber = tracing_subscriber::FmtSubscriber::new();
+    tracing::subscriber::set_global_default(subscriber)?;
+
+    let mut wallet_one = Wallet::create(RECV_ONE, CHANGE_ONE)
+        .network(NETWORK)
+        .create_wallet_no_persist()?;
+
+    let mut wallet_two = Wallet::create(RECV_TWO, CHANGE_TWO)
+        .network(NETWORK)
+        .create_wallet_no_persist()?;
+
+    // Build a request to sync for each wallet and place them in a vector.
+    let wallet_iter = vec![(&wallet_one, ScanType::Sync), (&wallet_two, ScanType::Sync)];
+
+    // Now build a client that will sync both wallets simultaneously
+    let client = Builder::new(NETWORK)
+        .build_with_wallets(wallet_iter)
+        .unwrap();
+    let (client, logging, mut update_subscriber) = client.subscribe();
+    tokio::task::spawn(
+        async move { traces(logging.info_subscriber, logging.warning_subscriber).await },
+    );
+    let client = client.start();
+    let requester = client.requester();
+
+    // Sync and apply updates. We can do this in a continual loop while the "application" is running.
+    // Often this would occur on a separate thread than the underlying application user interface.
+    loop {
+        // Updates are grouped with the `DescriptorId` of the public, external descriptor.
+        let updates = update_subscriber.updates().await?;
+        for (desc_id, update) in updates {
+            if wallet_one
+                .public_descriptor(KeychainKind::External)
+                .descriptor_id()
+                .eq(&desc_id)
+            {
+                wallet_one.apply_update(update)?;
+                tracing::info!("Wallet one summary: ");
+                tracing::info!("Balance: {:#}", wallet_one.balance().total());
+                tracing::info!(
+                    "Local chain tip: {}",
+                    wallet_one.local_chain().tip().height()
+                );
+            } else if wallet_two
+                .public_descriptor(KeychainKind::External)
+                .descriptor_id()
+                .eq(&desc_id)
+            {
+                wallet_two.apply_update(update)?;
+                tracing::info!("Wallet two summary: ");
+                tracing::info!("Balance: {:#}", wallet_two.balance().total());
+                tracing::info!(
+                    "Local chain tip: {}",
+                    wallet_two.local_chain().tip().height()
+                );
+            }
+        }
+        let fee_filter = requester.broadcast_min_feerate().await.unwrap();
+        tracing::info!("Broadcast minimum fee rate: {:#}", fee_filter);
+        tracing::info!("Press CTRL + C to exit.");
+    }
+}

src/builder.rs

@@ -47,8 +47,11 @@
 use std::fmt::Display;
 
 use bdk_wallet::{
-    chain::{CheckPoint, IndexedTxGraph},
-    Wallet,
+    chain::{
+        keychain_txout::KeychainTxOutIndex, CheckPoint, ConfirmationBlockTime, DescriptorExt,
+        DescriptorId, IndexedTxGraph,
+    },
+    KeychainKind, Wallet,
 };
 pub use bip157::Builder;
 use bip157::{chain::ChainState, HeaderCheckpoint};
@@ -64,15 +67,21 @@ pub trait BuilderExt {
         self,
         wallet: &Wallet,
         scan_type: ScanType,
-    ) -> Result<LightClient<Idle>, BuilderError>;
+    ) -> Result<LightClient<Idle, crate::wallets::Single>, BuilderError>;
+
+    /// Attempt to build the node with scripts from multiple [`Wallet`]s and following a [`ScanType`].
+    fn build_with_wallets(
+        self,
+        wallets: Vec<(&Wallet, ScanType)>,
+    ) -> Result<LightClient<Idle, crate::wallets::Multiple>, BuilderError>;
 }
 
 impl BuilderExt for Builder {
     fn build_with_wallet(
         mut self,
         wallet: &Wallet,
         scan_type: ScanType,
-    ) -> Result<LightClient<Idle>, BuilderError> {
+    ) -> Result<LightClient<Idle, crate::wallets::Single>, BuilderError> {
         let network = wallet.network();
         if self.network().ne(&network) {
             return Err(BuilderError::NetworkMismatch);
@@ -96,7 +105,7 @@ impl BuilderExt for Builder {
             event_rx,
         } = client;
         let indexed_graph = IndexedTxGraph::new(wallet.spk_index().clone());
-        let update_subscriber = UpdateSubscriber::new(
+        let update_subscriber = UpdateSubscriber::<crate::wallets::Single>::new(
             requester.clone(),
             scan_type,
             event_rx,
@@ -114,6 +123,72 @@ impl BuilderExt for Builder {
         );
         Ok(client)
     }
+
+    fn build_with_wallets(
+        mut self,
+        wallets: Vec<(&Wallet, ScanType)>,
+    ) -> Result<LightClient<Idle, crate::wallets::Multiple>, BuilderError> {
+        let network = wallets
+            .first()
+            .ok_or(BuilderError::EmptyIterator)?
+            .0
+            .network();
+        if self.network().ne(&network) {
+            return Err(BuilderError::NetworkMismatch);
+        }
+        let cp_min = wallets
+            .iter()
+            .map(|(wallet, scan_type)| match scan_type {
+                ScanType::Sync => walk_back_max_reorg(wallet.latest_checkpoint()),
+                ScanType::Recovery {
+                    used_script_index: _,
+                    checkpoint,
+                } => *checkpoint,
+            })
+            .min()
+            .ok_or(BuilderError::EmptyIterator)?;
+        self = self.chain_state(ChainState::Checkpoint(cp_min));
+        let (node, client) = self.build();
+        let bip157::Client {
+            requester,
+            info_rx,
+            warn_rx,
+            event_rx,
+        } = client;
+        let wallet_iter = wallets
+            .into_iter()
+            .map(|(wallet, scan_type)| {
+                (
+                    wallet
+                        .public_descriptor(KeychainKind::External)
+                        .descriptor_id(),
+                    scan_type,
+                    wallet.latest_checkpoint(),
+                    IndexedTxGraph::new(wallet.spk_index().clone()),
+                )
+            })
+            .collect::<Vec<(
+                DescriptorId,
+                ScanType,
+                CheckPoint,
+                IndexedTxGraph<ConfirmationBlockTime, KeychainTxOutIndex<KeychainKind>>,
+            )>>();
+        let update_subscriber = UpdateSubscriber::<crate::wallets::Multiple>::new_multiple(
+            requester.clone(),
+            event_rx,
+            wallet_iter.into_iter(),
+        );
+        let client = LightClient::new(
+            requester,
+            LoggingSubscribers {
+                info_subscriber: info_rx,
+                warning_subscriber: warn_rx,
+            },
+            update_subscriber,
+            node,
+        );
+        Ok(client)
+    }
 }
 
 /// Walk back 7 blocks in case the last sync was an orphan block.
@@ -134,6 +209,8 @@ fn walk_back_max_reorg(checkpoint: CheckPoint) -> HeaderCheckpoint {
 pub enum BuilderError {
     /// The wallet network and node network do not match.
     NetworkMismatch,
+    /// The wallet iterator is empty.
+    EmptyIterator,
 }
 
 impl Display for BuilderError {
@@ -142,6 +219,7 @@ impl Display for BuilderError {
             BuilderError::NetworkMismatch => {
                 write!(f, "wallet network and node network do not match")
             }
+            BuilderError::EmptyIterator => write!(f, "empty wallet iterator."),
         }
     }
 }