Implement deferred monitor write queueing and flushing

Replace the unimplemented!() stubs with a full deferred write
implementation. When ChainMonitor has deferred=true, Watch trait
operations queue PendingMonitorOp entries instead of executing
immediately. A new flush() method drains the queue and forwards
operations to the internal watch/update methods, calling
channel_monitor_updated on Completed status.

The BackgroundProcessor is updated to capture pending_operation_count
before persisting the ChannelManager, then flush that many writes
afterward - ensuring monitor writes happen in the correct order
relative to manager persistence.

Key changes:
- Add PendingMonitorOp enum and pending_ops queue to ChainMonitor
- Implement flush() and pending_operation_count() public methods
- Integrate flush calls in BackgroundProcessor (both sync and async)
- Add TestChainMonitor::new_deferred, flush helpers, and auto-flush
  in release_pending_monitor_events for test compatibility
- Add create_node_cfgs_deferred for deferred-mode test networks
- Add unit tests for queue/flush mechanics and full payment flow

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

Author: joostjager

lightning-background-processor/src/lib.rs

@@ -1120,6 +1120,10 @@ where
 
 		let mut futures = Joiner::new();
 
+		// Capture the pending count before persisting. Only this many writes will be
+		// flushed afterward, so that updates arriving after persist aren't included.
+		let pending_monitor_writes = chain_monitor.get_cm().pending_operation_count();
+
 		if channel_manager.get_cm().get_and_clear_needs_persistence() {
 			log_trace!(logger, "Persisting ChannelManager...");
 
@@ -1317,6 +1321,10 @@ where
 			res?;
 		}
 
+		// Flush monitor writes that were pending before we persisted. New updates that
+		// arrived after are left for the next iteration.
+		chain_monitor.get_cm().flush(pending_monitor_writes, &logger);
+
 		match check_and_reset_sleeper(&mut last_onion_message_handler_call, || {
 			sleeper(ONION_MESSAGE_HANDLER_TIMER)
 		}) {
@@ -1373,6 +1381,7 @@ where
 	// After we exit, ensure we persist the ChannelManager one final time - this avoids
 	// some races where users quit while channel updates were in-flight, with
 	// ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
+	let pending_monitor_writes = chain_monitor.get_cm().pending_operation_count();
 	kv_store
 		.write(
 			CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
@@ -1381,6 +1390,10 @@ where
 			channel_manager.get_cm().encode(),
 		)
 		.await?;
+
+	// Flush monitor writes that were pending before final persistence.
+	chain_monitor.get_cm().flush(pending_monitor_writes, &logger);
+
 	if let Some(ref scorer) = scorer {
 		kv_store
 			.write(
@@ -1571,7 +1584,8 @@ impl BackgroundProcessor {
 	) -> Self
 	where
 		L::Target: 'static + Logger,
-		M::Target: AChainMonitor<Signer = <CM::Target as AChannelManager>::Signer, Logger = L>,
+		M::Target:
+			'static + AChainMonitor<Signer = <CM::Target as AChannelManager>::Signer, Logger = L>,
 		CM::Target: AChannelManager,
 		OM::Target: AOnionMessenger,
 		PM::Target: APeerManager,
@@ -1684,6 +1698,11 @@ impl BackgroundProcessor {
 					channel_manager.get_cm().timer_tick_occurred();
 					last_freshness_call = Instant::now();
 				}
+
+				// Capture the pending count before persisting. Only this many writes will be
+				// flushed afterward, so that updates arriving after persist aren't included.
+				let pending_monitor_writes = chain_monitor.get_cm().pending_operation_count();
+
 				if channel_manager.get_cm().get_and_clear_needs_persistence() {
 					log_trace!(logger, "Persisting ChannelManager...");
 					(kv_store.write(
@@ -1695,6 +1714,10 @@ impl BackgroundProcessor {
 					log_trace!(logger, "Done persisting ChannelManager.");
 				}
 
+				// Flush monitor writes that were pending before we persisted. New updates
+				// that arrived after are left for the next iteration.
+				chain_monitor.get_cm().flush(pending_monitor_writes, &logger);
+
 				if let Some(liquidity_manager) = liquidity_manager.as_ref() {
 					log_trace!(logger, "Persisting LiquidityManager...");
 					let _ = liquidity_manager.get_lm().persist().map_err(|e| {
@@ -1809,12 +1832,17 @@ impl BackgroundProcessor {
 			// After we exit, ensure we persist the ChannelManager one final time - this avoids
 			// some races where users quit while channel updates were in-flight, with
 			// ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
+			let pending_monitor_writes = chain_monitor.get_cm().pending_operation_count();
 			kv_store.write(
 				CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
 				CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
 				CHANNEL_MANAGER_PERSISTENCE_KEY,
 				channel_manager.get_cm().encode(),
 			)?;
+
+			// Flush monitor writes that were pending before final persistence.
+			chain_monitor.get_cm().flush(pending_monitor_writes, &logger);
+
 			if let Some(ref scorer) = scorer {
 				kv_store.write(
 					SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
@@ -1896,9 +1924,10 @@ mod tests {
 	use bitcoin::transaction::{Transaction, TxOut};
 	use bitcoin::{Amount, ScriptBuf, Txid};
 	use core::sync::atomic::{AtomicBool, Ordering};
+	use lightning::chain::chainmonitor::ChainMonitor as LdkChainMonitor;
 	use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 	use lightning::chain::transaction::OutPoint;
-	use lightning::chain::{chainmonitor, BestBlock, Confirm, Filter};
+	use lightning::chain::{BestBlock, Confirm, Filter};
 	use lightning::events::{Event, PathFailure, ReplayEvent};
 	use lightning::ln::channelmanager;
 	use lightning::ln::channelmanager::{
@@ -2008,7 +2037,7 @@ mod tests {
 		Arc<test_utils::TestLogger>,
 	>;
 
-	type ChainMonitor = chainmonitor::ChainMonitor<
+	type ChainMonitor = LdkChainMonitor<
 		InMemorySigner,
 		Arc<test_utils::TestChainSource>,
 		Arc<test_utils::TestBroadcaster>,
@@ -2436,14 +2465,15 @@ 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(LdkChainMonitor::new(
 				Some(Arc::clone(&chain_source)),
 				Arc::clone(&tx_broadcaster),
 				Arc::clone(&logger),
 				Arc::clone(&fee_estimator),
 				Arc::clone(&kv_store),
 				Arc::clone(&keys_manager),
 				keys_manager.get_peer_storage_key(),
+				true,
 			));
 			let best_block = BestBlock::from_network(network);
 			let params = ChainParameters { network, best_block };
@@ -2580,19 +2610,31 @@ mod tests {
 					tx.clone(),
 				)
 				.unwrap();
+			// Flush deferred monitor operations so messages aren't held back
+			$node_a
+				.chain_monitor
+				.flush($node_a.chain_monitor.pending_operation_count(), &$node_a.logger);
 			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($node_b.chain_monitor.pending_operation_count(), &$node_b.logger);
 			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($node_a.chain_monitor.pending_operation_count(), &$node_a.logger);
 			get_event!($node_a, Event::ChannelPending);
 			tx
 		}};
@@ -3039,11 +3081,23 @@ 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(nodes[0].chain_monitor.pending_operation_count(), &nodes[0].logger);
 		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(nodes[1].chain_monitor.pending_operation_count(), &nodes[1].logger);
 		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(nodes[0].chain_monitor.pending_operation_count(), &nodes[0].logger);
 		channel_pending_recv
 			.recv_timeout(EVENT_DEADLINE)
 			.expect("ChannelPending not handled within deadline");
@@ -3104,6 +3158,10 @@ 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(nodes[0].chain_monitor.pending_operation_count(), &nodes[0].logger);
 		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);