Add DeferredChainMonitor wrapper for batched monitor persistence

Introduce DeferredChainMonitor, a wrapper around ChainMonitor that
queues watch_channel and update_channel operations, returning InProgress
until flush() is called. This enables batched persistence of monitor
updates after ChannelManager persistence, ensuring correct ordering.

The wrapper implements all public traits that ChainMonitor supports
(Listen, Confirm, EventsProvider, etc.) as pass-throughs, allowing it
to be used as a drop-in replacement.

Includes comprehensive tests covering the full channel lifecycle with
payment flows using DeferredChainMonitor.

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

Author: joostjager

lightning-background-processor/src/lib.rs

@@ -32,7 +32,8 @@ use fwd_batch::BatchDelay;
 
 use lightning::chain;
 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
-use lightning::chain::chainmonitor::{ChainMonitor, Persist};
+use lightning::chain::chainmonitor::Persist;
+use lightning::chain::deferred::DeferredChainMonitor;
 #[cfg(feature = "std")]
 use lightning::events::EventHandler;
 #[cfg(feature = "std")]
@@ -853,7 +854,7 @@ use futures_util::{dummy_waker, Joiner, OptionalSelector, Selector, SelectorOutp
 /// #     fn send_data(&mut self, _data: &[u8], _continue_read: bool) -> usize { 0 }
 /// #     fn disconnect_socket(&mut self) {}
 /// # }
-/// # type ChainMonitor<B, F, FE> = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<F>, Arc<B>, Arc<FE>, Arc<Logger>, Arc<StoreSync>, Arc<lightning::sign::KeysManager>>;
+/// # type ChainMonitor<B, F, FE> = lightning::chain::deferred::DeferredChainMonitor<lightning::sign::InMemorySigner, Arc<F>, Arc<B>, Arc<FE>, Arc<Logger>, Arc<StoreSync>, Arc<lightning::sign::KeysManager>>;
 /// # type NetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<Logger>>;
 /// # type P2PGossipSync<UL> = lightning::routing::gossip::P2PGossipSync<Arc<NetworkGraph>, Arc<UL>, Arc<Logger>>;
 /// # type ChannelManager<B, F, FE> = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor<B, F, FE>, B, FE, Logger>;
@@ -964,7 +965,9 @@ pub async fn process_events_async<
 	EventHandlerFuture: core::future::Future<Output = Result<(), ReplayEvent>>,
 	EventHandler: Fn(Event) -> EventHandlerFuture,
 	ES: Deref,
-	M: Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>>,
+	M: Deref<
+		Target = DeferredChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>,
+	>,
 	CM: Deref,
 	OM: Deref,
 	PGS: Deref<Target = P2PGossipSync<G, UL, L>>,
@@ -1155,7 +1158,7 @@ where
 		// Capture the number of pending monitor writes before persisting the channel manager.
 		// We'll only flush this many writes after the manager is persisted, to avoid flushing
 		// monitor updates that arrived after the manager state was captured.
-		let pending_monitor_writes = chain_monitor.pending_monitor_operation_count();
+		let pending_monitor_writes = chain_monitor.pending_operation_count();
 
 		if channel_manager.get_cm().get_and_clear_needs_persistence() {
 			log_trace!(logger, "Persisting ChannelManager...");
@@ -1427,7 +1430,7 @@ where
 		.await?;
 
 	// Flush all pending monitor writes after final channel manager persistence.
-	let pending_monitor_writes = chain_monitor.pending_monitor_operation_count();
+	let pending_monitor_writes = chain_monitor.pending_operation_count();
 	if pending_monitor_writes > 0 {
 		log_trace!(logger, "Flushing {} monitor writes on shutdown", pending_monitor_writes);
 		chain_monitor.flush(pending_monitor_writes);
@@ -1485,7 +1488,9 @@ pub async fn process_events_async_with_kv_store_sync<
 	EventHandlerFuture: core::future::Future<Output = Result<(), ReplayEvent>>,
 	EventHandler: Fn(Event) -> EventHandlerFuture,
 	ES: Deref,
-	M: Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>>,
+	M: Deref<
+		Target = DeferredChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>,
+	>,
 	CM: Deref,
 	OM: Deref,
 	PGS: Deref<Target = P2PGossipSync<G, UL, L>>,
@@ -1600,7 +1605,15 @@ impl BackgroundProcessor {
 		ES: 'static + Deref + Send,
 		M: 'static
 			+ Deref<
-				Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>,
+				Target = DeferredChainMonitor<
+					<CM::Target as AChannelManager>::Signer,
+					CF,
+					T,
+					F,
+					L,
+					P,
+					ES,
+				>,
 			>
 			+ Send
 			+ Sync,
@@ -1744,7 +1757,7 @@ impl BackgroundProcessor {
 				}
 
 				// Capture the number of pending monitor writes before persisting the channel manager.
-				let pending_monitor_writes = chain_monitor.pending_monitor_operation_count();
+				let pending_monitor_writes = chain_monitor.pending_operation_count();
 
 				if channel_manager.get_cm().get_and_clear_needs_persistence() {
 					log_trace!(logger, "Persisting ChannelManager...");
@@ -1885,7 +1898,7 @@ impl BackgroundProcessor {
 			)?;
 
 			// Flush all pending monitor writes after final channel manager persistence.
-			let pending_monitor_writes = chain_monitor.pending_monitor_operation_count();
+			let pending_monitor_writes = chain_monitor.pending_operation_count();
 			if pending_monitor_writes > 0 {
 				log_trace!(
 					logger,
@@ -1978,7 +1991,7 @@ mod tests {
 	use core::sync::atomic::{AtomicBool, Ordering};
 	use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 	use lightning::chain::transaction::OutPoint;
-	use lightning::chain::{chainmonitor, BestBlock, Confirm, Filter};
+	use lightning::chain::{deferred, BestBlock, Confirm, Filter};
 	use lightning::events::{Event, PathFailure, ReplayEvent};
 	use lightning::ln::channelmanager;
 	use lightning::ln::channelmanager::{
@@ -2068,7 +2081,7 @@ mod tests {
 		Arc<test_utils::TestLogger>,
 	>;
 
-	type ChainMonitor = chainmonitor::ChainMonitor<
+	type ChainMonitor = deferred::DeferredChainMonitor<
 		InMemorySigner,
 		Arc<test_utils::TestChainSource>,
 		Arc<test_utils::TestBroadcaster>,
@@ -2496,7 +2509,7 @@ mod tests {
 			let now = Duration::from_secs(genesis_block.header.time as u64);
 			let keys_manager =
 				Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos(), true));
-			let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(
+			let chain_monitor = Arc::new(deferred::DeferredChainMonitor::new(
 				Some(Arc::clone(&chain_source)),
 				Arc::clone(&tx_broadcaster),
 				Arc::clone(&logger),
@@ -2640,19 +2653,25 @@ mod tests {
 					tx.clone(),
 				)
 				.unwrap();
+			// Flush deferred monitor operations so messages aren't held back
+			$node_a.chain_monitor.flush_all();
 			let msg_a = get_event_msg!(
 				$node_a,
 				MessageSendEvent::SendFundingCreated,
 				$node_b.node.get_our_node_id()
 			);
 			$node_b.node.handle_funding_created($node_a.node.get_our_node_id(), &msg_a);
+			// Flush node_b's monitor so it releases the FundingSigned message
+			$node_b.chain_monitor.flush_all();
 			get_event!($node_b, Event::ChannelPending);
 			let msg_b = get_event_msg!(
 				$node_b,
 				MessageSendEvent::SendFundingSigned,
 				$node_a.node.get_our_node_id()
 			);
 			$node_a.node.handle_funding_signed($node_b.node.get_our_node_id(), &msg_b);
+			// Flush node_a's monitor for the final update
+			$node_a.chain_monitor.flush_all();
 			get_event!($node_a, Event::ChannelPending);
 			tx
 		}};
@@ -3099,11 +3118,17 @@ mod tests {
 			.node
 			.funding_transaction_generated(temporary_channel_id, node_1_id, funding_tx.clone())
 			.unwrap();
+		// Flush node_0's deferred monitor operations so the FundingCreated message is released
+		nodes[0].chain_monitor.flush_all();
 		let msg_0 = get_event_msg!(nodes[0], MessageSendEvent::SendFundingCreated, node_1_id);
 		nodes[1].node.handle_funding_created(node_0_id, &msg_0);
+		// Flush node_1's deferred monitor operations so events and FundingSigned are released
+		nodes[1].chain_monitor.flush_all();
 		get_event!(nodes[1], Event::ChannelPending);
 		let msg_1 = get_event_msg!(nodes[1], MessageSendEvent::SendFundingSigned, node_0_id);
 		nodes[0].node.handle_funding_signed(node_1_id, &msg_1);
+		// Flush node_0's monitor for the funding_signed update
+		nodes[0].chain_monitor.flush_all();
 		channel_pending_recv
 			.recv_timeout(EVENT_DEADLINE)
 			.expect("ChannelPending not handled within deadline");
@@ -3164,6 +3189,8 @@ mod tests {
 				error_message.to_string(),
 			)
 			.unwrap();
+		// Flush the monitor update triggered by force close so the commitment tx is broadcasted
+		nodes[0].chain_monitor.flush_all();
 		let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
 		confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);