Testing the new potential kyoto client

bdk-android/build.gradle.kts

@@ -1,6 +1,6 @@
 plugins {
     id("com.android.library").version("8.13.2").apply(false)
-    id("org.jetbrains.kotlin.android").version("2.1.10").apply(false)
+    id("org.jetbrains.kotlin.android").version("2.3.0").apply(false)
     id("org.jetbrains.dokka").version("2.1.0").apply(false)
     id("com.vanniktech.maven.publish").version("0.36.0").apply(false)
 }

bdk-android/lib/build.gradle.kts

@@ -56,6 +56,7 @@ dependencies {
     androidTestImplementation("androidx.test.espresso:espresso-core:3.4.0")
     androidTestImplementation("org.jetbrains.kotlin:kotlin-test:1.6.10")
     androidTestImplementation("org.jetbrains.kotlin:kotlin-test-junit:1.6.10")
+    androidTestImplementation("org.kotlinbitcointools:regtest-toolbox:0.2.0")
 }
 
 mavenPublishing {

bdk-android/lib/src/androidTest/kotlin/org/bitcoindevkit/CbfSyncTest.kt

@@ -0,0 +1,79 @@
+package org.bitcoindevkit
+
+import kotlin.test.Test
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import androidx.test.platform.app.InstrumentationRegistry
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.junit.runner.RunWith
+// import org.kotlinbitcointools.regtesttoolbox.regenv.RegEnv
+import java.io.File
+import kotlin.time.Duration.Companion.seconds
+
+@RunWith(AndroidJUnit4::class)
+class CbfSyncTest {
+    private val conn: Persister = Persister.newInMemory()
+
+    private val kyotoDataDir: String by lazy {
+        val context = InstrumentationRegistry.getInstrumentation().targetContext
+        val dir = File(context.filesDir, "kyoto_test_data")
+        dir.mkdirs()
+        dir.absolutePath
+    }
+
+    @Test
+    fun syncWithKyotoClient() {
+        runBlocking {
+            // println("Hello, Compact Block Filters!")
+            // val regtestEnv = RegEnv.connectTo(walletName = "faucet", username = "regtest", password = "password")
+
+            val wallet: Wallet = Wallet.createSingle(
+                descriptor = BIP86_DESCRIPTOR,
+                network = Network.REGTEST,
+                persister = conn
+            )
+            val newAddress = wallet.revealNextAddress(KeychainKind.EXTERNAL).address
+            println("New address: $newAddress")
+            delay(2.seconds)
+
+            // regtestEnv.send(newAddress.toString(), 0.12345678, 2.0)
+            // regtestEnv.mine(2)
+
+            val ip: IpAddress = IpAddress.fromIpv4(10u, 0u, 2u, 2u)
+            val peer1: Peer = Peer(ip, 18444u, false)
+            val peers: List<Peer> = listOf(peer1)
+            val node = CbfBuilder()
+                .peers(peers)
+                .connections(1u)
+                .scanType(ScanType.Sync)
+                .dataDir(kyotoDataDir)
+                .build(wallet)
+
+            println("Test is complete")
+            val client = node.start()
+
+            val warningJob = launch {
+                try {
+                    while (true) println("Warning: ${client.nextWarning()}")
+                } catch (e: CbfException) { println("Warning channel closed: $e") }
+            }
+            val infoJob = launch {
+                try {
+                    while (true) println("Info: ${client.nextInfo()}")
+                } catch (e: CbfException) { println("Info channel closed: $e") }
+            }
+
+            val update: Update = client.update()
+            delay(8.seconds)
+            wallet.applyUpdate(update)
+
+            val balance = wallet.balance().total.toSat()
+            assert(balance > 0uL)
+
+            warningJob.cancel()
+            infoJob.cancel()
+            client.shutdown()
+        }
+    }
+}

bdk-ffi/Cargo.toml

@@ -18,7 +18,8 @@ path = "uniffi-bindgen.rs"
 bdk_wallet = { version = "2.3.0", features = ["all-keys", "keys-bip39", "rusqlite"] }
 bdk_esplora = { version = "0.22.1", default-features = false, features = ["std", "blocking", "blocking-https-rustls"] }
 bdk_electrum = { version = "0.23.2", default-features = false, features = ["use-rustls-ring"] }
-bdk_kyoto = { version = "0.15.4" }
+bdk_kyoto = { git = "https://github.com/bitcoindevkit/bdk-kyoto.git", branch = "master" }
+#bdk_kyoto = { version = "0.15.4" }
 
 uniffi = { version = "=0.30.0", features = ["cli"]}
 thiserror = "2.0.17"

bdk-ffi/src/error.rs

@@ -797,6 +797,8 @@ pub enum TxidParseError {
 pub enum CbfError {
     #[error("the node is no longer running")]
     NodeStopped,
+    #[error("the node has already been started")]
+    AlreadyStarted,
 }
 
 // ------------------------------------------------------------------------

bdk-ffi/src/kyoto.rs

@@ -2,19 +2,17 @@ use bdk_kyoto::bip157::lookup_host;
 use bdk_kyoto::bip157::tokio;
 use bdk_kyoto::bip157::AddrV2;
 use bdk_kyoto::bip157::Network;
-use bdk_kyoto::bip157::Node;
 use bdk_kyoto::bip157::ServiceFlags;
 use bdk_kyoto::builder::Builder as BDKCbfBuilder;
 use bdk_kyoto::builder::BuilderExt;
-use bdk_kyoto::HeaderCheckpoint;
-use bdk_kyoto::LightClient as BDKLightClient;
 use bdk_kyoto::Receiver;
 use bdk_kyoto::RejectReason;
 use bdk_kyoto::Requester;
 use bdk_kyoto::TrustedPeer;
 use bdk_kyoto::UnboundedReceiver;
 use bdk_kyoto::UpdateSubscriber;
 use bdk_kyoto::Warning as Warn;
+use bdk_kyoto::{HeaderCheckpoint, Idle, LightClient};
 
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 use std::path::PathBuf;
@@ -37,51 +35,37 @@ const CWD_PATH: &str = ".";
 const TCP_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(2);
 const MESSAGE_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
 
-/// Receive a [`CbfClient`] and [`CbfNode`].
-#[derive(Debug, uniffi::Record)]
-pub struct CbfComponents {
-    /// Publish events to the node, like broadcasting transactions or adding scripts.
-    pub client: Arc<CbfClient>,
-    /// The node to run and fetch transactions for a [`Wallet`].
-    pub node: Arc<CbfNode>,
+/// A compact block filters node that has been configured but not yet started.
+///
+/// Built via [`CbfBuilder`]. Call [`CbfNode::start`] to prepare and spawn the node, receiving a
+/// [`CbfClient`] to interact with the running node.
+///
+/// `CbfNode::start` may only be called once. Calling it a second time returns
+/// [`CbfError::AlreadyStarted`].
+#[derive(uniffi::Object)]
+pub struct CbfNode {
+    client: std::sync::Mutex<Option<LightClient<Idle>>>,
 }
 
-/// A [`CbfClient`] handles wallet updates from a [`CbfNode`].
-#[derive(Debug, uniffi::Object)]
+/// Interact with a running compact block filters node.
+///
+/// Obtained by calling [`CbfNode::start`]. Provides access to three independent
+/// channels — each message is consumed by exactly one caller, so dedicate a single
+/// task or thread to draining each channel:
+///
+/// * [`CbfClient::next_info`] — progress and connection events.
+/// * [`CbfClient::next_warning`] — non-fatal issues the node encountered.
+/// * [`CbfClient::update`] — wallet updates ready to be applied to a [`Wallet`].
+///
+/// Transactions can be broadcast and peers added at any time via the remaining methods.
+#[derive(uniffi::Object)]
 pub struct CbfClient {
-    sender: Arc<Requester>,
+    sender: Requester,
     info_rx: Mutex<Receiver<bdk_kyoto::Info>>,
     warning_rx: Mutex<UnboundedReceiver<bdk_kyoto::Warning>>,
     update_rx: Mutex<UpdateSubscriber>,
 }
 
-/// A [`CbfNode`] gathers transactions for a [`Wallet`].
-/// To receive [`Update`] for [`Wallet`], refer to the
-/// [`CbfClient`]. The [`CbfNode`] will run until instructed
-/// to stop.
-#[derive(Debug, uniffi::Object)]
-pub struct CbfNode {
-    node: std::sync::Mutex<Option<Node>>,
-}
-
-#[uniffi::export]
-impl CbfNode {
-    /// Start the node on a detached OS thread and immediately return.
-    pub fn run(self: Arc<Self>) {
-        let mut lock = self.node.lock().unwrap();
-        let node = lock.take().expect("cannot call run more than once");
-        std::thread::spawn(|| {
-            tokio::runtime::Builder::new_multi_thread()
-                .enable_all()
-                .build()
-                .unwrap()
-                .block_on(async move {
-                    let _ = node.run().await;
-                })
-        });
-    }
-}
-
 /// Build a BIP 157/158 light client to fetch transactions for a `Wallet`.
 ///
 /// Options:
@@ -178,7 +162,7 @@ impl CbfBuilder {
     }
 
     /// Construct a [`CbfComponents`] for a [`Wallet`].
-    pub fn build(&self, wallet: &Wallet) -> CbfComponents {
+    pub fn build(&self, wallet: &Wallet) -> CbfNode {
         let wallet = wallet.get_wallet();
 
         let mut trusted_peers = Vec::new();
@@ -241,31 +225,51 @@ impl CbfBuilder {
             builder = builder.socks5_proxy((addr, port));
         }
 
-        let BDKLightClient {
-            requester,
-            info_subscriber,
-            warning_subscriber,
-            update_subscriber,
-            node,
-        } = builder
+        let light_client_idle = builder
             .build_with_wallet(&wallet, scan_type)
             .expect("networks match by definition");
 
-        let node = CbfNode {
-            node: std::sync::Mutex::new(Some(node)),
-        };
+        CbfNode {
+            client: std::sync::Mutex::new(Some(light_client_idle)),
+        }
+    }
+}
 
-        let client = CbfClient {
-            sender: Arc::new(requester),
-            info_rx: Mutex::new(info_subscriber),
-            warning_rx: Mutex::new(warning_subscriber),
-            update_rx: Mutex::new(update_subscriber),
-        };
+#[uniffi::export]
+impl CbfNode {
+    /// Subscribe to log and update channels, then spawn the node on a background thread.
+    ///
+    /// Returns a [`CbfClient`] that can be used to receive wallet updates, info and warning
+    /// messages, and to broadcast transactions.
+    ///
+    /// This method may only be called once. A second call returns [`CbfError::AlreadyStarted`].
+    pub fn start(&self) -> Result<Arc<CbfClient>, CbfError> {
+        let light_client = self
+            .client
+            .lock()
+            .unwrap()
+            .take()
+            .ok_or(CbfError::AlreadyStarted)?;
 
-        CbfComponents {
-            client: Arc::new(client),
-            node: Arc::new(node),
-        }
+        let (subscribed, logging, updates) = light_client.subscribe();
+        let (active, node) = subscribed.managed_start();
+
+        std::thread::spawn(|| {
+            tokio::runtime::Builder::new_multi_thread()
+                .enable_all()
+                .build()
+                .unwrap()
+                .block_on(async move {
+                    let _ = node.run().await;
+                })
+        });
+
+        Ok(Arc::new(CbfClient {
+            sender: active.requester(),
+            info_rx: Mutex::new(logging.info_subscriber),
+            warning_rx: Mutex::new(logging.warning_subscriber),
+            update_rx: Mutex::new(updates),
+        }))
     }
 }
 
@@ -340,7 +344,7 @@ impl CbfClient {
     /// Add another [`Peer`] to attempt a connection with.
     pub fn connect(&self, peer: Peer) -> Result<(), CbfError> {
         self.sender
-            .add_peer(peer)
+            .add_peer(TrustedPeer::from(peer))
             .map_err(|_| CbfError::NodeStopped)
     }