Expose async monitor persistence for tests

Downstream tests need to exercise the same async monitor persistence
path used by ChainMonitor::new_async_beta without reimplementing
Persist.
Add test-only constructors that let TestChainMonitor and AsyncPersister
share the wake notifier so async completions drive the monitor update
future.

Co-Authored-By: HAL 9000

Author: tnull

lightning/src/chain/chainmonitor.rs

@@ -278,6 +278,31 @@ pub struct AsyncPersister<
 	event_notifier: Arc<Notifier>,
 }
 
+#[cfg(feature = "_test_utils")]
+impl<
+		K: KVStore + MaybeSend + MaybeSync + 'static,
+		S: FutureSpawner,
+		L: Logger + MaybeSend + MaybeSync + 'static,
+		ES: EntropySource + MaybeSend + MaybeSync + 'static,
+		SP: SignerProvider + MaybeSend + MaybeSync + 'static,
+		BI: BroadcasterInterface + MaybeSend + MaybeSync + 'static,
+		FE: FeeEstimator + MaybeSend + MaybeSync + 'static,
+	> AsyncPersister<K, S, L, ES, SP, BI, FE>
+where
+	SP::EcdsaSigner: MaybeSend + 'static,
+{
+	/// Constructs a test-only [`AsyncPersister`] with the given event notifier.
+	///
+	/// The same notifier should be passed to the [`ChainMonitor`] which consumes this persister so
+	/// that async persistence completions wake the monitor's update future.
+	pub fn new_test(
+		persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, BI, FE>,
+		event_notifier: Arc<Notifier>,
+	) -> Self {
+		Self { persister, event_notifier }
+	}
+}
+
 impl<
 		K: KVStore + MaybeSend + MaybeSync + 'static,
 		S: FutureSpawner,
@@ -648,6 +673,43 @@ where
 	pub fn new(
 		chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, persister: P,
 		_entropy_source: ES, _our_peerstorage_encryption_key: PeerStorageKey, deferred: bool,
+	) -> Self {
+		Self::with_event_notifier(
+			chain_source,
+			broadcaster,
+			logger,
+			feeest,
+			persister,
+			_entropy_source,
+			_our_peerstorage_encryption_key,
+			deferred,
+			Arc::new(Notifier::new()),
+		)
+	}
+
+	#[cfg(feature = "_test_utils")]
+	pub(crate) fn new_with_event_notifier(
+		chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, persister: P,
+		_entropy_source: ES, _our_peerstorage_encryption_key: PeerStorageKey, deferred: bool,
+		event_notifier: Arc<Notifier>,
+	) -> Self {
+		Self::with_event_notifier(
+			chain_source,
+			broadcaster,
+			logger,
+			feeest,
+			persister,
+			_entropy_source,
+			_our_peerstorage_encryption_key,
+			deferred,
+			event_notifier,
+		)
+	}
+
+	fn with_event_notifier(
+		chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, persister: P,
+		_entropy_source: ES, _our_peerstorage_encryption_key: PeerStorageKey, deferred: bool,
+		event_notifier: Arc<Notifier>,
 	) -> Self {
 		Self {
 			monitors: RwLock::new(new_hash_map()),
@@ -659,7 +721,7 @@ where
 			_entropy_source,
 			pending_monitor_events: Mutex::new(Vec::new()),
 			highest_chain_height: AtomicUsize::new(0),
-			event_notifier: Arc::new(Notifier::new()),
+			event_notifier,
 			pending_send_only_events: Mutex::new(Vec::new()),
 			#[cfg(peer_storage)]
 			our_peerstorage_encryption_key: _our_peerstorage_encryption_key,

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -4924,6 +4924,50 @@ fn test_single_channel_multiple_mpp() {
 	check_added_monitors(&nodes[7], 1);
 }
 
+#[cfg(feature = "_test_utils")]
+#[test]
+fn async_persister_new_test_works_with_test_chain_monitor() {
+	let logger = Arc::new(test_utils::TestLogger::new());
+	let keys_manager = Arc::new(test_utils::TestKeysInterface::new(&[42; 32], Network::Testnet));
+	let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(Network::Testnet));
+	let fee_estimator = Arc::new(test_utils::TestFeeEstimator::new(253));
+	let kv_store = Arc::new(test_utils::TestStore::new(false));
+	let persist_futures = Arc::new(FutureQueue::new());
+	let native_async_persister = MonitorUpdatingPersisterAsync::new(
+		Arc::clone(&kv_store),
+		Arc::clone(&persist_futures),
+		Arc::clone(&logger),
+		42,
+		Arc::clone(&keys_manager),
+		Arc::clone(&keys_manager),
+		Arc::clone(&tx_broadcaster),
+		Arc::clone(&fee_estimator),
+	);
+	let event_notifier = Arc::new(crate::util::wakers::Notifier::new());
+	let async_persister = crate::chain::chainmonitor::AsyncPersister::new_test(
+		native_async_persister,
+		Arc::clone(&event_notifier),
+	);
+	assert_eq!(Arc::strong_count(&event_notifier), 2);
+
+	let chain_source = test_utils::TestChainSource::new(Network::Testnet);
+	let test_chain_monitor = test_utils::TestChainMonitor::new_with_event_notifier(
+		Some(&chain_source),
+		tx_broadcaster.as_ref(),
+		logger.as_ref(),
+		fee_estimator.as_ref(),
+		&async_persister,
+		keys_manager.as_ref(),
+		Arc::clone(&event_notifier),
+	);
+	assert_eq!(Arc::strong_count(&event_notifier), 3);
+
+	let update_future = test_chain_monitor.chain_monitor.get_update_future();
+	assert!(!update_future.poll_is_complete());
+	event_notifier.notify();
+	assert!(update_future.poll_is_complete());
+}
+
 #[test]
 fn native_async_persist() {
 	// Test ChainMonitor::new_async_beta and the backing MonitorUpdatingPersisterAsync.

lightning/src/util/test_utils.rs

@@ -555,6 +555,25 @@ impl<'a> TestChainMonitor<'a> {
 		)
 	}
 
+	#[cfg(feature = "_test_utils")]
+	pub fn new_with_event_notifier(
+		chain_source: Option<&'a TestChainSource>, broadcaster: &'a dyn SyncBroadcaster,
+		logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator,
+		persister: &'a dyn SyncPersist, keys_manager: &'a TestKeysInterface,
+		event_notifier: Arc<Notifier>,
+	) -> Self {
+		Self::with_deferred_and_event_notifier(
+			chain_source,
+			broadcaster,
+			logger,
+			fee_estimator,
+			persister,
+			keys_manager,
+			false,
+			Some(event_notifier),
+		)
+	}
+
 	pub fn new_deferred(
 		chain_source: Option<&'a TestChainSource>, broadcaster: &'a dyn SyncBroadcaster,
 		logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator,
@@ -576,11 +595,27 @@ impl<'a> TestChainMonitor<'a> {
 		logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator,
 		persister: &'a dyn SyncPersist, keys_manager: &'a TestKeysInterface, deferred: bool,
 	) -> Self {
-		Self {
-			added_monitors: Mutex::new(Vec::new()),
-			monitor_updates: Mutex::new(new_hash_map()),
-			latest_monitor_update_id: Mutex::new(new_hash_map()),
-			chain_monitor: ChainMonitor::new(
+		Self::with_deferred_and_event_notifier(
+			chain_source,
+			broadcaster,
+			logger,
+			fee_estimator,
+			persister,
+			keys_manager,
+			deferred,
+			None,
+		)
+	}
+
+	fn with_deferred_and_event_notifier(
+		chain_source: Option<&'a TestChainSource>, broadcaster: &'a dyn SyncBroadcaster,
+		logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator,
+		persister: &'a dyn SyncPersist, keys_manager: &'a TestKeysInterface, deferred: bool,
+		event_notifier: Option<Arc<Notifier>>,
+	) -> Self {
+		#[cfg(feature = "_test_utils")]
+		let chain_monitor = if let Some(event_notifier) = event_notifier {
+			ChainMonitor::new_with_event_notifier(
 				chain_source,
 				broadcaster,
 				logger,
@@ -589,7 +624,39 @@ impl<'a> TestChainMonitor<'a> {
 				keys_manager,
 				keys_manager.get_peer_storage_key(),
 				deferred,
-			),
+				event_notifier,
+			)
+		} else {
+			ChainMonitor::new(
+				chain_source,
+				broadcaster,
+				logger,
+				fee_estimator,
+				persister,
+				keys_manager,
+				keys_manager.get_peer_storage_key(),
+				deferred,
+			)
+		};
+		#[cfg(not(feature = "_test_utils"))]
+		let chain_monitor = {
+			let _ = event_notifier;
+			ChainMonitor::new(
+				chain_source,
+				broadcaster,
+				logger,
+				fee_estimator,
+				persister,
+				keys_manager,
+				keys_manager.get_peer_storage_key(),
+				deferred,
+			)
+		};
+		Self {
+			added_monitors: Mutex::new(Vec::new()),
+			monitor_updates: Mutex::new(new_hash_map()),
+			latest_monitor_update_id: Mutex::new(new_hash_map()),
+			chain_monitor,
 			keys_manager,
 			logger,
 			expect_channel_force_closed: Mutex::new(None),