Use async chain monitor persister

Use `ChainMonitor::new_async_beta` with `MonitorUpdatingPersisterAsync`
for chain monitor persistence.

Add `DynStoreRef`, a newtype wrapper that bridges the object-safe
`DynStoreTrait` (boxed futures) to LDK's generic `KVStore` trait
(`impl Future`), as required by `MonitorUpdatingPersisterAsync`.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: joostjager

src/builder.rs

@@ -76,9 +76,9 @@ use crate::peer_store::PeerStore;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
-	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, GossipSync, Graph,
-	KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore,
-	Persister, SyncAndAsyncKVStore,
+	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper,
+	GossipSync, Graph, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager,
+	PendingPaymentStore, SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -1495,7 +1495,7 @@ fn build_with_store_internal(
 
 	let peer_storage_key = keys_manager.get_peer_storage_key();
 	let monitor_reader = Arc::new(AsyncPersister::new(
-		Arc::clone(&kv_store),
+		DynStoreRef(Arc::clone(&kv_store)),
 		RuntimeSpawner::new(Arc::clone(&runtime)),
 		Arc::clone(&logger),
 		PERSISTER_MAX_PENDING_UPDATES,
@@ -1508,7 +1508,7 @@ fn build_with_store_internal(
 	// Read ChannelMonitors and the NetworkGraph
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (monitor_read_res, network_graph_res) = runtime.block_on(async move {
+	let (monitor_read_res, network_graph_res) = runtime.block_on(async {
 		tokio::join!(
 			monitor_reader.read_all_channel_monitors_with_updates_parallel(),
 			read_network_graph(&*kv_store_ref, logger_ref),
@@ -1528,27 +1528,21 @@ fn build_with_store_internal(
 		},
 	};
 
-	let persister = Arc::new(Persister::new(
-		Arc::clone(&kv_store),
-		Arc::clone(&logger),
-		PERSISTER_MAX_PENDING_UPDATES,
-		Arc::clone(&keys_manager),
-		Arc::clone(&keys_manager),
-		Arc::clone(&tx_broadcaster),
-		Arc::clone(&fee_estimator),
-	));
-
 	// Initialize the ChainMonitor
-	let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(
-		Some(Arc::clone(&chain_source)),
-		Arc::clone(&tx_broadcaster),
-		Arc::clone(&logger),
-		Arc::clone(&fee_estimator),
-		Arc::clone(&persister),
-		Arc::clone(&keys_manager),
-		peer_storage_key,
-		true,
-	));
+	let chain_monitor: Arc<ChainMonitor> = {
+		let persister = Arc::try_unwrap(monitor_reader)
+			.unwrap_or_else(|_| panic!("Arc<AsyncPersister> should have no other references"));
+		Arc::new(chainmonitor::ChainMonitor::new_async_beta(
+			Some(Arc::clone(&chain_source)),
+			Arc::clone(&tx_broadcaster),
+			Arc::clone(&logger),
+			Arc::clone(&fee_estimator),
+			persister,
+			Arc::clone(&keys_manager),
+			peer_storage_key,
+			true,
+		))
+	};
 
 	// Initialize the network graph, scorer, and router
 	let network_graph = match network_graph_res {

src/types.rs

@@ -23,9 +23,7 @@ use lightning::routing::gossip;
 use lightning::routing::router::DefaultRouter;
 use lightning::routing::scoring::{CombinedScorer, ProbabilisticScoringFeeParameters};
 use lightning::sign::InMemorySigner;
-use lightning::util::persist::{
-	KVStore, KVStoreSync, MonitorUpdatingPersister, MonitorUpdatingPersisterAsync,
-};
+use lightning::util::persist::{KVStore, KVStoreSync, MonitorUpdatingPersisterAsync};
 use lightning::util::ser::{Readable, Writeable, Writer};
 use lightning::util::sweep::OutputSweeper;
 use lightning_block_sync::gossip::GossipVerifier;
@@ -135,6 +133,39 @@ impl<'a> KVStoreSync for dyn DynStoreTrait + 'a {
 
 pub(crate) type DynStore = dyn DynStoreTrait;
 
+// Newtype wrapper that implements `KVStore` for `Arc<DynStore>`. This is needed because `KVStore`
+// methods return `impl Future`, which is not object-safe. `DynStoreTrait` works around this by
+// returning `Pin<Box<dyn Future>>` instead, and this wrapper bridges the two by delegating
+// `KVStore` methods to the corresponding `DynStoreTrait::*_async` methods.
+#[derive(Clone)]
+pub(crate) struct DynStoreRef(pub(crate) Arc<DynStore>);
+
+impl KVStore for DynStoreRef {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> impl Future<Output = Result<Vec<u8>, bitcoin::io::Error>> + Send + 'static {
+		DynStoreTrait::read_async(&*self.0, primary_namespace, secondary_namespace, key)
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> impl Future<Output = Result<(), bitcoin::io::Error>> + Send + 'static {
+		DynStoreTrait::write_async(&*self.0, primary_namespace, secondary_namespace, key, buf)
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> impl Future<Output = Result<(), bitcoin::io::Error>> + Send + 'static {
+		DynStoreTrait::remove_async(&*self.0, primary_namespace, secondary_namespace, key, lazy)
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> impl Future<Output = Result<Vec<String>, bitcoin::io::Error>> + Send + 'static {
+		DynStoreTrait::list_async(&*self.0, primary_namespace, secondary_namespace)
+	}
+}
+
 pub(crate) struct DynStoreWrapper<T: SyncAndAsyncKVStore + Send + Sync>(pub(crate) T);
 
 impl<T: SyncAndAsyncKVStore + Send + Sync> DynStoreTrait for DynStoreWrapper<T> {
@@ -188,7 +219,7 @@ impl<T: SyncAndAsyncKVStore + Send + Sync> DynStoreTrait for DynStoreWrapper<T>
 }
 
 pub(crate) type AsyncPersister = MonitorUpdatingPersisterAsync<
-	Arc<DynStore>,
+	DynStoreRef,
 	RuntimeSpawner,
 	Arc<Logger>,
 	Arc<KeysManager>,
@@ -197,22 +228,21 @@ pub(crate) type AsyncPersister = MonitorUpdatingPersisterAsync<
 	Arc<OnchainFeeEstimator>,
 >;
 
-pub type Persister = MonitorUpdatingPersister<
-	Arc<DynStore>,
-	Arc<Logger>,
-	Arc<KeysManager>,
-	Arc<KeysManager>,
-	Arc<Broadcaster>,
-	Arc<OnchainFeeEstimator>,
->;
-
 pub(crate) type ChainMonitor = chainmonitor::ChainMonitor<
 	InMemorySigner,
 	Arc<ChainSource>,
 	Arc<Broadcaster>,
 	Arc<OnchainFeeEstimator>,
 	Arc<Logger>,
-	Arc<Persister>,
+	chainmonitor::AsyncPersister<
+		DynStoreRef,
+		RuntimeSpawner,
+		Arc<Logger>,
+		Arc<KeysManager>,
+		Arc<KeysManager>,
+		Arc<Broadcaster>,
+		Arc<OnchainFeeEstimator>,
+	>,
 	Arc<KeysManager>,
 >;