Fully switch to async KVStore persistence

.github/workflows/benchmarks.yml

@@ -43,4 +43,4 @@ jobs:
           echo "ELECTRS_EXE=$( pwd )/bin/electrs-${{ runner.os }}-${{ runner.arch }}" >> "$GITHUB_ENV"
       - name: Run benchmarks
         run: |
-          cargo bench
+          RUSTFLAGS="--cfg tokio_unstable" cargo bench

.github/workflows/rust.yml

@@ -84,7 +84,7 @@ jobs:
       - name: Test on Rust ${{ matrix.toolchain }}
         if: "matrix.platform != 'windows-latest'"
         run: |
-          RUSTFLAGS="--cfg no_download --cfg cycle_tests" cargo test
+          RUSTFLAGS="--cfg no_download --cfg cycle_tests --cfg tokio_unstable" cargo test
       - name: Test with UniFFI support on Rust ${{ matrix.toolchain }}
         if: "matrix.platform != 'windows-latest' && matrix.build-uniffi"
         run: |
@@ -114,4 +114,4 @@ jobs:
       - uses: actions/checkout@v6
       - uses: dtolnay/rust-toolchain@nightly
       - uses: dtolnay/install@cargo-docs-rs
-      - run: cargo docs-rs
+      - run: cargo docs-rs

Cargo.toml

@@ -196,3 +196,17 @@ harness = false
 #lightning-liquidity = { path = "../rust-lightning/lightning-liquidity" }
 #lightning-macros = { path = "../rust-lightning/lightning-macros" }
 #lightning-dns-resolver = { path = "../rust-lightning/lightning-dns-resolver" }
+
+[patch."https://github.com/lightningdevkit/rust-lightning"]
+lightning = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-types = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-invoice = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-net-tokio = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-persister = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-background-processor = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-rapid-gossip-sync = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-block-sync = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-transaction-sync = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-liquidity = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-macros = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }
+lightning-dns-resolver = { git = "https://github.com/tnull/rust-lightning", rev = "2611766e6a5913d6e33afdc1932485575ee8ea4e" }

scripts/uniffi_bindgen_generate_kotlin.sh

@@ -5,6 +5,11 @@ PROJECT_DIR="ldk-node-jvm"
 PACKAGE_DIR="org/lightningdevkit/ldknode"
 UNIFFI_BINDGEN_BIN="cargo run --manifest-path bindings/uniffi-bindgen/Cargo.toml"
 
+case " ${RUSTFLAGS:-} " in
+	*" --cfg tokio_unstable "*|*" --cfg=tokio_unstable "*) ;;
+	*) export RUSTFLAGS="${RUSTFLAGS:+$RUSTFLAGS }--cfg tokio_unstable" ;;
+esac
+
 if [[ "$OSTYPE" == "linux-gnu"* ]]; then
 	rustup target add x86_64-unknown-linux-gnu || exit 1
 	cargo build --release --target x86_64-unknown-linux-gnu --features uniffi || exit 1

scripts/uniffi_bindgen_generate_kotlin_android.sh

@@ -5,6 +5,11 @@ TARGET_DIR="target"
 PROJECT_DIR="ldk-node-android"
 UNIFFI_BINDGEN_BIN="cargo run --manifest-path bindings/uniffi-bindgen/Cargo.toml"
 
+case " ${RUSTFLAGS:-} " in
+	*" --cfg tokio_unstable "*|*" --cfg=tokio_unstable "*) RUSTFLAGS_WITH_TOKIO_UNSTABLE="${RUSTFLAGS:-}" ;;
+	*) RUSTFLAGS_WITH_TOKIO_UNSTABLE="${RUSTFLAGS:+$RUSTFLAGS }--cfg tokio_unstable" ;;
+esac
+
 export_variable_if_not_present() {
   local name="$1"
   local value="$2"
@@ -35,9 +40,9 @@ case "$OSTYPE" in
 PATH="$ANDROID_NDK_ROOT/toolchains/llvm/prebuilt/$LLVM_ARCH_PATH/bin:$PATH"
 
 rustup target add x86_64-linux-android aarch64-linux-android armv7-linux-androideabi
-RUSTFLAGS="-C link-args=-Wl,-z,max-page-size=16384,-z,common-page-size=16384" CFLAGS="-D__ANDROID_MIN_SDK_VERSION__=21" AR=llvm-ar CARGO_TARGET_X86_64_LINUX_ANDROID_LINKER="x86_64-linux-android21-clang" CC="x86_64-linux-android21-clang" cargo build --profile release-smaller --features uniffi --target x86_64-linux-android || exit 1
-RUSTFLAGS="-C link-args=-Wl,-z,max-page-size=16384,-z,common-page-size=16384" CFLAGS="-D__ANDROID_MIN_SDK_VERSION__=21" AR=llvm-ar CARGO_TARGET_ARMV7_LINUX_ANDROIDEABI_LINKER="armv7a-linux-androideabi21-clang" CC="armv7a-linux-androideabi21-clang" cargo build --profile release-smaller --features uniffi --target armv7-linux-androideabi || exit 1
-RUSTFLAGS="-C link-args=-Wl,-z,max-page-size=16384,-z,common-page-size=16384" CFLAGS="-D__ANDROID_MIN_SDK_VERSION__=21" AR=llvm-ar CARGO_TARGET_AARCH64_LINUX_ANDROID_LINKER="aarch64-linux-android21-clang" CC="aarch64-linux-android21-clang" cargo build --profile release-smaller --features uniffi --target aarch64-linux-android || exit 1
+RUSTFLAGS="$RUSTFLAGS_WITH_TOKIO_UNSTABLE -C link-args=-Wl,-z,max-page-size=16384,-z,common-page-size=16384" CFLAGS="-D__ANDROID_MIN_SDK_VERSION__=21" AR=llvm-ar CARGO_TARGET_X86_64_LINUX_ANDROID_LINKER="x86_64-linux-android21-clang" CC="x86_64-linux-android21-clang" cargo build --profile release-smaller --features uniffi --target x86_64-linux-android || exit 1
+RUSTFLAGS="$RUSTFLAGS_WITH_TOKIO_UNSTABLE -C link-args=-Wl,-z,max-page-size=16384,-z,common-page-size=16384" CFLAGS="-D__ANDROID_MIN_SDK_VERSION__=21" AR=llvm-ar CARGO_TARGET_ARMV7_LINUX_ANDROIDEABI_LINKER="armv7a-linux-androideabi21-clang" CC="armv7a-linux-androideabi21-clang" cargo build --profile release-smaller --features uniffi --target armv7-linux-androideabi || exit 1
+RUSTFLAGS="$RUSTFLAGS_WITH_TOKIO_UNSTABLE -C link-args=-Wl,-z,max-page-size=16384,-z,common-page-size=16384" CFLAGS="-D__ANDROID_MIN_SDK_VERSION__=21" AR=llvm-ar CARGO_TARGET_AARCH64_LINUX_ANDROID_LINKER="aarch64-linux-android21-clang" CC="aarch64-linux-android21-clang" cargo build --profile release-smaller --features uniffi --target aarch64-linux-android || exit 1
 $UNIFFI_BINDGEN_BIN generate bindings/ldk_node.udl --lib-file "$TARGET_DIR"/x86_64-linux-android/release-smaller/libldk_node.so --language kotlin --config uniffi-android.toml -o "$BINDINGS_DIR"/"$PROJECT_DIR"/lib/src/main/kotlin || exit 1
 
 JNI_LIB_DIR="$BINDINGS_DIR"/"$PROJECT_DIR"/lib/src/main/jniLibs/ 

scripts/uniffi_bindgen_generate_python.sh

@@ -2,6 +2,11 @@
 BINDINGS_DIR="./bindings/python/src/ldk_node"
 UNIFFI_BINDGEN_BIN="cargo run --manifest-path bindings/uniffi-bindgen/Cargo.toml"
 
+case " ${RUSTFLAGS:-} " in
+	*" --cfg tokio_unstable "*|*" --cfg=tokio_unstable "*) ;;
+	*) export RUSTFLAGS="${RUSTFLAGS:+$RUSTFLAGS }--cfg tokio_unstable" ;;
+esac
+
 if [[ "$OSTYPE" == "linux-gnu"* ]]; then
 	DYNAMIC_LIB_PATH="./target/release-smaller/libldk_node.so"
 else

scripts/uniffi_bindgen_generate_swift.sh

@@ -4,6 +4,11 @@ set -eox pipefail
 BINDINGS_DIR="./bindings/swift"
 UNIFFI_BINDGEN_BIN="cargo run --manifest-path bindings/uniffi-bindgen/Cargo.toml"
 
+case " ${RUSTFLAGS:-} " in
+	*" --cfg tokio_unstable "*|*" --cfg=tokio_unstable "*) ;;
+	*) export RUSTFLAGS="${RUSTFLAGS:+$RUSTFLAGS }--cfg tokio_unstable" ;;
+esac
+
 mkdir -p $BINDINGS_DIR
 
 # Install rust target toolchains

src/builder.rs

@@ -81,11 +81,11 @@ use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
 	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper,
 	GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, PaymentStore,
-	PeerManager, PendingPaymentStore, SyncAndAsyncKVStore,
+	PeerManager, PendingPaymentStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
-use crate::{Node, NodeMetrics};
+use crate::{Node, NodeMetrics, PersistedNodeMetrics};
 
 const LSPS_HARDENED_CHILD_INDEX: u32 = 577;
 const PERSISTER_MAX_PENDING_UPDATES: u64 = 100;
@@ -176,17 +176,17 @@ pub enum BuildError {
 	RuntimeSetupFailed,
 	/// We failed to read data from the [`KVStore`].
 	///
-	/// [`KVStore`]: lightning::util::persist::KVStoreSync
+	/// [`KVStore`]: lightning::util::persist::KVStore
 	ReadFailed,
 	/// We failed to write data to the [`KVStore`].
 	///
-	/// [`KVStore`]: lightning::util::persist::KVStoreSync
+	/// [`KVStore`]: lightning::util::persist::KVStore
 	WriteFailed,
 	/// We failed to access the given `storage_dir_path`.
 	StoragePathAccessFailed,
 	/// We failed to setup our [`KVStore`].
 	///
-	/// [`KVStore`]: lightning::util::persist::KVStoreSync
+	/// [`KVStore`]: lightning::util::persist::KVStore
 	KVStoreSetupFailed,
 	/// We failed to setup the onchain wallet.
 	WalletSetupFailed,
@@ -697,11 +697,12 @@ impl NodeBuilder {
 	/// [`FilesystemStoreV2`]: lightning_persister::fs_store::v2::FilesystemStoreV2
 	pub fn build_with_fs_store(&self, node_entropy: NodeEntropy) -> Result<Node, BuildError> {
 		let logger = setup_logger(&self.log_writer_config, &self.config)?;
+		let runtime = self.setup_runtime(&logger)?;
 		let mut storage_dir_path: PathBuf = self.config.storage_dir_path.clone().into();
 		storage_dir_path.push("fs_store");
 
-		let kv_store = open_or_migrate_fs_store(storage_dir_path)?;
-		self.build_with_store_and_logger(node_entropy, kv_store, logger)
+		let kv_store = runtime.block_on(open_or_migrate_fs_store(storage_dir_path))?;
+		self.build_with_store_runtime_and_logger(node_entropy, kv_store, runtime, logger)
 	}
 
 	/// Builds a [`Node`] instance with a [VSS] backend and according to the options
@@ -825,7 +826,7 @@ impl NodeBuilder {
 	}
 
 	/// Builds a [`Node`] instance according to the options previously configured.
-	pub fn build_with_store<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
+	pub fn build_with_store<S: KVStore + Send + Sync + 'static>(
 		&self, node_entropy: NodeEntropy, kv_store: S,
 	) -> Result<Node, BuildError> {
 		let logger = setup_logger(&self.log_writer_config, &self.config)?;
@@ -844,14 +845,14 @@ impl NodeBuilder {
 		}
 	}
 
-	fn build_with_store_and_logger<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
+	fn build_with_store_and_logger<S: KVStore + Send + Sync + 'static>(
 		&self, node_entropy: NodeEntropy, kv_store: S, logger: Arc<Logger>,
 	) -> Result<Node, BuildError> {
 		let runtime = self.setup_runtime(&logger)?;
 		self.build_with_store_runtime_and_logger(node_entropy, kv_store, runtime, logger)
 	}
 
-	fn build_with_store_runtime_and_logger<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
+	fn build_with_store_runtime_and_logger<S: KVStore + Send + Sync + 'static>(
 		&self, node_entropy: NodeEntropy, kv_store: S, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Result<Node, BuildError> {
 		let seed_bytes = node_entropy.to_seed_bytes();
@@ -1345,7 +1346,7 @@ impl ArcedNodeBuilder {
 	/// Builds a [`Node`] instance according to the options previously configured.
 	// Note that the generics here don't actually work for Uniffi, but we don't currently expose
 	// this so its not needed.
-	pub fn build_with_store<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
+	pub fn build_with_store<S: KVStore + Send + Sync + 'static>(
 		&self, node_entropy: Arc<NodeEntropy>, kv_store: S,
 	) -> Result<Arc<Node>, BuildError> {
 		self.inner.read().expect("lock").build_with_store(*node_entropy, kv_store).map(Arc::new)
@@ -1415,10 +1416,10 @@ fn build_with_store_internal(
 
 	// Initialize the status fields.
 	let node_metrics = match node_metris_res {
-		Ok(metrics) => Arc::new(RwLock::new(metrics)),
+		Ok(metrics) => Arc::new(PersistedNodeMetrics::new(metrics)),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
-				Arc::new(RwLock::new(NodeMetrics::default()))
+				Arc::new(PersistedNodeMetrics::new(NodeMetrics::default()))
 			} else {
 				log_error!(logger, "Failed to read node metrics from store: {}", e);
 				return Err(BuildError::ReadFailed);
@@ -1539,12 +1540,16 @@ fn build_with_store_internal(
 	let change_descriptor = Bip84(xprv, KeychainKind::Internal);
 	let mut wallet_persister =
 		KVStoreWalletPersister::new(Arc::clone(&kv_store), Arc::clone(&logger));
-	let wallet_opt = BdkWallet::load()
-		.descriptor(KeychainKind::External, Some(descriptor.clone()))
-		.descriptor(KeychainKind::Internal, Some(change_descriptor.clone()))
-		.extract_keys()
-		.check_network(config.network)
-		.load_wallet(&mut wallet_persister)
+	let wallet_opt = runtime
+		.block_on(async {
+			BdkWallet::load()
+				.descriptor(KeychainKind::External, Some(descriptor.clone()))
+				.descriptor(KeychainKind::Internal, Some(change_descriptor.clone()))
+				.extract_keys()
+				.check_network(config.network)
+				.load_wallet_async(&mut wallet_persister)
+				.await
+		})
 		.map_err(|e| match e {
 			bdk_wallet::LoadWithPersistError::InvalidChangeSet(
 				bdk_wallet::LoadError::Mismatch(bdk_wallet::LoadMismatch::Network {
@@ -1568,9 +1573,13 @@ fn build_with_store_internal(
 	let bdk_wallet = match wallet_opt {
 		Some(wallet) => wallet,
 		None => {
-			let mut wallet = BdkWallet::create(descriptor, change_descriptor)
-				.network(config.network)
-				.create_wallet(&mut wallet_persister)
+			let mut wallet = runtime
+				.block_on(async {
+					BdkWallet::create(descriptor, change_descriptor)
+						.network(config.network)
+						.create_wallet_async(&mut wallet_persister)
+						.await
+				})
 				.map_err(|e| {
 					log_error!(logger, "Failed to set up wallet: {}", e);
 					BuildError::WalletSetupFailed
@@ -1619,6 +1628,7 @@ fn build_with_store_internal(
 		Arc::clone(&fee_estimator),
 		Arc::clone(&chain_source),
 		Arc::clone(&payment_store),
+		Arc::clone(&runtime),
 		Arc::clone(&config),
 		Arc::clone(&logger),
 		Arc::clone(&pending_payment_store),

src/chain/bitcoind.rs

@@ -42,7 +42,7 @@ use crate::fee_estimator::{
 use crate::io::utils::update_and_persist_node_metrics;
 use crate::logger::{log_bytes, log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
-use crate::{Error, NodeMetrics};
+use crate::{Error, PersistedNodeMetrics};
 
 const CHAIN_POLLING_INTERVAL_SECS: u64 = 2;
 const CHAIN_POLLING_TIMEOUT_SECS: u64 = 10;
@@ -55,14 +55,14 @@ pub(super) struct BitcoindChainSource {
 	kv_store: Arc<DynStore>,
 	config: Arc<Config>,
 	logger: Arc<Logger>,
-	node_metrics: Arc<RwLock<NodeMetrics>>,
+	node_metrics: Arc<PersistedNodeMetrics>,
 }
 
 impl BitcoindChainSource {
 	pub(crate) fn new_rpc(
 		rpc_host: String, rpc_port: u16, rpc_user: String, rpc_password: String,
 		fee_estimator: Arc<OnchainFeeEstimator>, kv_store: Arc<DynStore>, config: Arc<Config>,
-		logger: Arc<Logger>, node_metrics: Arc<RwLock<NodeMetrics>>,
+		logger: Arc<Logger>, node_metrics: Arc<PersistedNodeMetrics>,
 	) -> Self {
 		let api_client = Arc::new(BitcoindClient::new_rpc(
 			rpc_host.clone(),
@@ -89,7 +89,7 @@ impl BitcoindChainSource {
 		rpc_host: String, rpc_port: u16, rpc_user: String, rpc_password: String,
 		fee_estimator: Arc<OnchainFeeEstimator>, kv_store: Arc<DynStore>, config: Arc<Config>,
 		rest_client_config: BitcoindRestClientConfig, logger: Arc<Logger>,
-		node_metrics: Arc<RwLock<NodeMetrics>>,
+		node_metrics: Arc<PersistedNodeMetrics>,
 	) -> Self {
 		let api_client = Arc::new(BitcoindClient::new_rest(
 			rest_client_config.rest_host,
@@ -204,6 +204,7 @@ impl BitcoindChainSource {
 								m.latest_onchain_wallet_sync_timestamp = unix_time_secs_opt;
 							},
 						)
+						.await
 						.unwrap_or_else(|e| {
 							log_error!(self.logger, "Failed to persist node metrics: {}", e);
 						});
@@ -451,7 +452,8 @@ impl BitcoindChainSource {
 		update_and_persist_node_metrics(&self.node_metrics, &*self.kv_store, &*self.logger, |m| {
 			m.latest_lightning_wallet_sync_timestamp = unix_time_secs_opt;
 			m.latest_onchain_wallet_sync_timestamp = unix_time_secs_opt;
-		})?;
+		})
+		.await?;
 
 		Ok(())
 	}
@@ -563,7 +565,8 @@ impl BitcoindChainSource {
 			SystemTime::now().duration_since(UNIX_EPOCH).ok().map(|d| d.as_secs());
 		update_and_persist_node_metrics(&self.node_metrics, &*self.kv_store, &*self.logger, |m| {
 			m.latest_fee_rate_cache_update_timestamp = unix_time_secs_opt
-		})?;
+		})
+		.await?;
 
 		Ok(())
 	}