Implement deferred monitor write queueing and flushing

diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
index fc58eda..594681d 100644
--- a/lightning-background-processor/src/lib.rs
+++ b/lightning-background-processor/src/lib.rs
@@ -773,6 +773,17 @@ use futures_util::{dummy_waker, Joiner, OptionalSelector, Selector, SelectorOutp
 /// The `fetch_time` parameter should return the current wall clock time, if one is available. If
 /// no time is available, some features may be disabled, however the node will still operate fine.
 ///
+/// Note that when deferred monitor writes are enabled on [`ChainMonitor`], this function flushes
+/// pending writes after persisting the [`ChannelManager`]. If the [`Persist`] implementation
+/// performs blocking I/O and returns [`Completed`] synchronously rather than returning
+/// [`InProgress`], this will block the async executor.
+///
+/// [`ChainMonitor`]: lightning::chain::chainmonitor::ChainMonitor
+/// [`Persist`]: lightning::chain::chainmonitor::Persist
+/// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
+/// [`Completed`]: lightning::chain::ChannelMonitorUpdateStatus::Completed
+/// [`InProgress`]: lightning::chain::ChannelMonitorUpdateStatus::InProgress
+///
 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
 /// could setup `process_events_async` like this:
 /// ```
@@ -1116,9 +1127,18 @@ where
 			None => {},
 		}
 
+		// We capture pending_operation_count inside the persistence branch to
+		// avoid a race: ChannelManager handlers queue deferred monitor ops
+		// before the persistence flag is set. Capturing outside would let us
+		// observe pending ops while the flag is still unset, causing us to
+		// flush monitor writes without persisting the ChannelManager.
+		// Declared before futures so it outlives the Joiner (drop order).
+		let pending_monitor_writes;
+
 		let mut futures = Joiner::new();
 
 		if channel_manager.get_cm().get_and_clear_needs_persistence() {
+			pending_monitor_writes = chain_monitor.get_cm().pending_operation_count();
 			log_trace!(logger, "Persisting ChannelManager...");
 
 			let fut = async {
@@ -1129,7 +1149,12 @@ where
 						CHANNEL_MANAGER_PERSISTENCE_KEY,
 						channel_manager.get_cm().encode(),
 					)
-					.await
+					.await?;
+
+				// Flush monitor operations 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);
+				Ok(())
 			};
 			// TODO: Once our MSRV is 1.68 we should be able to drop the Box
 			let mut fut = Box::pin(fut);
@@ -1371,6 +1396,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,
@@ -1379,6 +1405,10 @@ where
 			channel_manager.get_cm().encode(),
 		)
 		.await?;
+
+	// Flush monitor operations that were pending before final persistence.
+	chain_monitor.get_cm().flush(pending_monitor_writes, &logger);
+
 	if let Some(ref scorer) = scorer {
 		kv_store
 			.write(
@@ -1682,7 +1712,15 @@ impl BackgroundProcessor {
 					channel_manager.get_cm().timer_tick_occurred();
 					last_freshness_call = Instant::now();
 				}
+
 				if channel_manager.get_cm().get_and_clear_needs_persistence() {
+					// We capture pending_operation_count inside the persistence
+					// branch to avoid a race: ChannelManager handlers queue
+					// deferred monitor ops before the persistence flag is set.
+					// Capturing outside would let us observe pending ops while
+					// the flag is still unset, causing us to flush monitor
+					// writes without persisting the ChannelManager.
+					let pending_monitor_writes = chain_monitor.get_cm().pending_operation_count();
 					log_trace!(logger, "Persisting ChannelManager...");
 					(kv_store.write(
 						CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
@@ -1691,6 +1729,10 @@ impl BackgroundProcessor {
 						channel_manager.get_cm().encode(),
 					))?;
 					log_trace!(logger, "Done persisting ChannelManager.");
+
+					// Flush monitor operations 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() {
@@ -1807,12 +1849,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 operations 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,
@@ -1894,9 +1941,10 @@ mod tests {
 	use bitcoin::transaction::{Transaction, TxOut};
 	use bitcoin::{Amount, ScriptBuf, Txid};
 	use core::sync::atomic::{AtomicBool, Ordering};
+	use lightning::chain::chainmonitor;
 	use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 	use lightning::chain::transaction::OutPoint;
-	use lightning::chain::{chainmonitor, BestBlock, Confirm};
+	use lightning::chain::{BestBlock, Confirm};
 	use lightning::events::{Event, PathFailure, ReplayEvent};
 	use lightning::ln::channelmanager;
 	use lightning::ln::channelmanager::{
@@ -2441,6 +2489,7 @@ mod tests {
 				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 };
@@ -2562,6 +2611,8 @@ mod tests {
 		(persist_dir, nodes)
 	}
 
+	/// Opens a channel between two nodes without a running `BackgroundProcessor`,
+	/// so deferred monitor operations are flushed manually at each step.
 	macro_rules! open_channel {
 		($node_a: expr, $node_b: expr, $channel_value: expr) => {{
 			begin_open_channel!($node_a, $node_b, $channel_value);
@@ -2577,12 +2628,19 @@ mod tests {
 					tx.clone(),
 				)
 				.unwrap();
+			// funding_transaction_generated does not call watch_channel, so no
+			// deferred op is queued and FundingCreated is available immediately.
 			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 new monitor (watch_channel) 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,
@@ -2590,6 +2648,11 @@ mod tests {
 				$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 new monitor (watch_channel) queued by
+			// handle_funding_signed.
+			$node_a
+				.chain_monitor
+				.flush($node_a.chain_monitor.pending_operation_count(), &$node_a.logger);
 			get_event!($node_a, Event::ChannelPending);
 			tx
 		}};
@@ -2715,6 +2778,20 @@ mod tests {
 		confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
 	}
 
+	/// Waits until the background processor has flushed all pending deferred monitor
+	/// operations for the given node. Panics if the pending count does not reach zero
+	/// within `EVENT_DEADLINE`.
+	fn wait_for_flushed(chain_monitor: &ChainMonitor) {
+		let start = std::time::Instant::now();
+		while chain_monitor.pending_operation_count() > 0 {
+			assert!(
+				start.elapsed() < EVENT_DEADLINE,
+				"Pending monitor operations were not flushed within deadline"
+			);
+			std::thread::sleep(Duration::from_millis(10));
+		}
+	}
+
 	#[test]
 	fn test_background_processor() {
 		// Test that when a new channel is created, the ChannelManager needs to be re-persisted with
@@ -3055,11 +3132,21 @@ mod tests {
 			.node
 			.funding_transaction_generated(temporary_channel_id, node_1_id, funding_tx.clone())
 			.unwrap();
+		// funding_transaction_generated does not call watch_channel, so no deferred op is
+		// queued and the FundingCreated message is available immediately.
 		let msg_0 = get_event_msg!(nodes[0], MessageSendEvent::SendFundingCreated, node_1_id);
 		nodes[1].node.handle_funding_created(node_0_id, &msg_0);
+		// Node 1 has no bg processor, flush its new monitor (watch_channel) manually 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);
+		// Wait for the bg processor to flush the new monitor (watch_channel) queued by
+		// handle_funding_signed.
+		wait_for_flushed(&nodes[0].chain_monitor);
 		channel_pending_recv
 			.recv_timeout(EVENT_DEADLINE)
 			.expect("ChannelPending not handled within deadline");
@@ -3120,6 +3207,9 @@ mod tests {
 				error_message.to_string(),
 			)
 			.unwrap();
+		// Wait for the bg processor to flush the monitor update triggered by force close
+		// so the commitment tx is broadcast.
+		wait_for_flushed(&nodes[0].chain_monitor);
 		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);
 
diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs
index 99f792f..215b48e 100644
--- a/lightning/src/chain/chainmonitor.rs
+++ b/lightning/src/chain/chainmonitor.rs
@@ -60,12 +60,21 @@ use crate::util::persist::{KVStore, MonitorName, MonitorUpdatingPersisterAsync};
 use crate::util::ser::{VecWriter, Writeable};
 use crate::util::wakers::{Future, Notifier};
 
+use alloc::collections::VecDeque;
 use alloc::sync::Arc;
 #[cfg(peer_storage)]
 use core::iter::Cycle;
 use core::ops::Deref;
 use core::sync::atomic::{AtomicUsize, Ordering};
 
+/// A pending operation queued for later execution when `ChainMonitor` is in deferred mode.
+enum PendingMonitorOp<ChannelSigner: EcdsaChannelSigner> {
+	/// A new monitor to insert and persist.
+	NewMonitor { channel_id: ChannelId, monitor: ChannelMonitor<ChannelSigner> },
+	/// An update to apply and persist.
+	Update { channel_id: ChannelId, update: ChannelMonitorUpdate },
+}
+
 /// `Persist` defines behavior for persisting channel monitors: this could mean
 /// writing once to disk, and/or uploading to one or more backup services.
 ///
@@ -376,6 +385,10 @@ pub struct ChainMonitor<
 
 	/// When `true`, [`chain::Watch`] operations are queued rather than executed immediately.
 	deferred: bool,
+	/// Queued monitor operations awaiting flush. Unused when `deferred` is `false`.
+	pending_ops: Mutex<VecDeque<PendingMonitorOp<ChannelSigner>>>,
+	/// Guards [`Self::flush`] so that concurrent calls are serialized.
+	flush_lock: Mutex<()>,
 }
 
 impl<
@@ -398,6 +411,18 @@ where
 	///
 	/// Note that async monitor updating is considered beta, and bugs may be triggered by its use.
 	///
+	/// When `deferred` is `true`, [`chain::Watch::watch_channel`] and
+	/// [`chain::Watch::update_channel`] calls are not executed immediately. Instead, they are
+	/// queued internally and must be flushed by the caller via [`Self::flush`]. Use
+	/// [`Self::pending_operation_count`] to check how many operations are queued, then call
+	/// [`Self::flush`] to process them. This allows the caller to ensure that the
+	/// [`ChannelManager`] is persisted before its associated monitors, avoiding the risk of
+	/// force closures from a crash between monitor and channel manager persistence.
+	///
+	/// When `deferred` is `false`, monitor operations are executed inline as usual.
+	///
+	/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
+	///
 	/// This is not exported to bindings users as async is not supported outside of Rust.
 	pub fn new_async_beta(
 		chain_source: Option<C>, broadcaster: T, logger: L, feeest: F,
@@ -420,6 +445,8 @@ where
 			#[cfg(peer_storage)]
 			our_peerstorage_encryption_key: _our_peerstorage_encryption_key,
 			deferred,
+			pending_ops: Mutex::new(VecDeque::new()),
+			flush_lock: Mutex::new(()),
 		}
 	}
 }
@@ -604,6 +631,16 @@ where
 	/// is obtained by the [`ChannelManager`] through [`NodeSigner`] to decrypt peer backups.
 	/// Using an inconsistent or incorrect key will result in the inability to decrypt previously encrypted backups.
 	///
+	/// When `deferred` is `true`, [`chain::Watch::watch_channel`] and
+	/// [`chain::Watch::update_channel`] calls are not executed immediately. Instead, they are
+	/// queued internally and must be flushed by the caller via [`Self::flush`]. Use
+	/// [`Self::pending_operation_count`] to check how many operations are queued, then call
+	/// [`Self::flush`] to process them. This allows the caller to ensure that the
+	/// [`ChannelManager`] is persisted before its associated monitors, avoiding the risk of
+	/// force closures from a crash between monitor and channel manager persistence.
+	///
+	/// When `deferred` is `false`, monitor operations are executed inline as usual.
+	///
 	/// [`NodeSigner`]: crate::sign::NodeSigner
 	/// [`NodeSigner::get_peer_storage_key`]: crate::sign::NodeSigner::get_peer_storage_key
 	/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
@@ -626,6 +663,8 @@ where
 			#[cfg(peer_storage)]
 			our_peerstorage_encryption_key: _our_peerstorage_encryption_key,
 			deferred,
+			pending_ops: Mutex::new(VecDeque::new()),
+			flush_lock: Mutex::new(()),
 		}
 	}
 
@@ -1045,7 +1084,7 @@ where
 		&self, channel_id: ChannelId, monitor: ChannelMonitor<ChannelSigner>,
 	) -> Result<ChannelMonitorUpdateStatus, ()> {
 		if !monitor.written_by_0_1_or_later() {
-			return chain::Watch::watch_channel(self, channel_id, monitor);
+			return self.watch_channel_internal(channel_id, monitor);
 		}
 
 		let logger = WithChannelMonitor::from(&self.logger, &monitor, None);
@@ -1219,6 +1258,90 @@ where
 			},
 		}
 	}
+
+	/// Returns the number of pending monitor operations queued for later execution.
+	///
+	/// When the `ChainMonitor` is constructed with `deferred` set to `true`,
+	/// [`chain::Watch::watch_channel`] and [`chain::Watch::update_channel`] calls are queued
+	/// instead of being executed immediately. Call this method to determine how many operations
+	/// are waiting, then pass the result to [`Self::flush`] to process them.
+	pub fn pending_operation_count(&self) -> usize {
+		self.pending_ops.lock().unwrap().len()
+	}
+
+	/// Flushes the first `count` pending monitor operations that were queued while the
+	/// `ChainMonitor` operates in deferred mode. `count` must not exceed the number of
+	/// pending operations returned by [`Self::pending_operation_count`].
+	///
+	/// A typical usage pattern is to call [`Self::pending_operation_count`], persist the
+	/// [`ChannelManager`], then pass the count to this method to flush the queued operations.
+	///
+	/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
+	pub fn flush(&self, count: usize, logger: &L) {
+		let _guard = self.flush_lock.lock().unwrap();
+		if count > 0 {
+			log_info!(logger, "Flushing up to {} monitor operations", count);
+		}
+		for _ in 0..count {
+			let mut queue = self.pending_ops.lock().unwrap();
+			let op = match queue.pop_front() {
+				Some(op) => op,
+				None => {
+					debug_assert!(false, "flush count exceeded queue length");
+					return;
+				},
+			};
+
+			let (channel_id, update_id, status) = match op {
+				PendingMonitorOp::NewMonitor { channel_id, monitor } => {
+					let logger = WithChannelMonitor::from(logger, &monitor, None);
+					let update_id = monitor.get_latest_update_id();
+					log_trace!(logger, "Flushing new monitor");
+					// Hold `pending_ops` across the internal call so that
+					// `watch_channel` (which checks `monitors` + `pending_ops`
+					// atomically) cannot race with this insertion.
+					match self.watch_channel_internal(channel_id, monitor) {
+						Ok(status) => {
+							drop(queue);
+							(channel_id, update_id, status)
+						},
+						Err(()) => {
+							// `watch_channel` checks both `pending_ops` and `monitors`
+							// for duplicates before queueing, so this is unreachable.
+							unreachable!();
+						},
+					}
+				},
+				PendingMonitorOp::Update { channel_id, update } => {
+					let logger = WithContext::from(logger, None, Some(channel_id), None);
+					log_trace!(logger, "Flushing monitor update {}", update.update_id);
+					// Release `pending_ops` before the internal call so that
+					// concurrent `update_channel` queuing is not blocked.
+					drop(queue);
+					let update_id = update.update_id;
+					let status = self.update_channel_internal(channel_id, &update);
+					(channel_id, update_id, status)
+				},
+			};
+
+			match status {
+				ChannelMonitorUpdateStatus::Completed => {
+					let logger = WithContext::from(logger, None, Some(channel_id), None);
+					if let Err(e) = self.channel_monitor_updated(channel_id, update_id) {
+						debug_assert!(false, "channel_monitor_updated failed: {:?}", e);
+						log_error!(logger, "channel_monitor_updated failed: {:?}", e);
+					}
+				},
+				ChannelMonitorUpdateStatus::InProgress => {},
+				ChannelMonitorUpdateStatus::UnrecoverableError => {
+					// Neither watch_channel_internal nor update_channel_internal
+					// return UnrecoverableError; they panic on that variant
+					// before it can be returned.
+					unreachable!();
+				},
+			}
+		}
+	}
 }
 
 impl<
@@ -1437,7 +1560,22 @@ where
 			return self.watch_channel_internal(channel_id, monitor);
 		}
 
-		unimplemented!();
+		// Atomically check for duplicates in both the pending queue and the
+		// flushed monitor set.
+		let mut pending_ops = self.pending_ops.lock().unwrap();
+		let monitors = self.monitors.read().unwrap();
+		if monitors.contains_key(&channel_id) {
+			return Err(());
+		}
+		let already_pending = pending_ops.iter().any(|op| match op {
+			PendingMonitorOp::NewMonitor { channel_id: id, .. } => *id == channel_id,
+			_ => false,
+		});
+		if already_pending {
+			return Err(());
+		}
+		pending_ops.push_back(PendingMonitorOp::NewMonitor { channel_id, monitor });
+		Ok(ChannelMonitorUpdateStatus::InProgress)
 	}
 
 	fn update_channel(
@@ -1447,7 +1585,21 @@ where
 			return self.update_channel_internal(channel_id, update);
 		}
 
-		unimplemented!();
+		let mut pending_ops = self.pending_ops.lock().unwrap();
+		debug_assert!(
+			{
+				let monitors = self.monitors.read().unwrap();
+				let in_monitors = monitors.contains_key(&channel_id);
+				let in_pending = pending_ops.iter().any(|op| match op {
+					PendingMonitorOp::NewMonitor { channel_id: id, .. } => *id == channel_id,
+					_ => false,
+				});
+				in_monitors || in_pending
+			},
+			"ChannelManager generated a channel update for a channel that was not yet registered!"
+		);
+		pending_ops.push_back(PendingMonitorOp::Update { channel_id, update: update.clone() });
+		ChannelMonitorUpdateStatus::InProgress
 	}
 
 	fn release_pending_monitor_events(
@@ -1577,12 +1729,22 @@ where
 
 #[cfg(test)]
 mod tests {
-	use crate::chain::channelmonitor::ANTI_REORG_DELAY;
+	use super::ChainMonitor;
+	use crate::chain::channelmonitor::{ChannelMonitorUpdate, ANTI_REORG_DELAY};
 	use crate::chain::{ChannelMonitorUpdateStatus, Watch};
 	use crate::events::{ClosureReason, Event};
 	use crate::ln::functional_test_utils::*;
 	use crate::ln::msgs::{BaseMessageHandler, ChannelMessageHandler, MessageSendEvent};
+	use crate::ln::types::ChannelId;
+	use crate::sign::NodeSigner;
+	use crate::util::dyn_signer::DynSigner;
+	use crate::util::test_channel_signer::TestChannelSigner;
+	use crate::util::test_utils::{
+		TestBroadcaster, TestChainSource, TestFeeEstimator, TestKeysInterface, TestLogger,
+		TestPersister,
+	};
 	use crate::{expect_payment_path_successful, get_event_msg};
+	use bitcoin::Network;
 
 	const CHAINSYNC_MONITOR_PARTITION_FACTOR: u32 = 5;
 
@@ -1840,4 +2002,171 @@ mod tests {
 		})
 		.is_err());
 	}
+
+	/// Concrete `ChainMonitor` type wired to the standard test utilities in deferred mode.
+	type TestDeferredChainMonitor<'a> = ChainMonitor<
+		TestChannelSigner,
+		&'a TestChainSource,
+		&'a TestBroadcaster,
+		&'a TestFeeEstimator,
+		&'a TestLogger,
+		&'a TestPersister,
+		&'a TestKeysInterface,
+	>;
+
+	/// Creates a minimal `ChannelMonitorUpdate` with no actual update steps.
+	fn dummy_update(update_id: u64, channel_id: ChannelId) -> ChannelMonitorUpdate {
+		ChannelMonitorUpdate { updates: vec![], update_id, channel_id: Some(channel_id) }
+	}
+
+	fn create_deferred_chain_monitor<'a>(
+		chain_source: &'a TestChainSource, broadcaster: &'a TestBroadcaster,
+		logger: &'a TestLogger, fee_est: &'a TestFeeEstimator, persister: &'a TestPersister,
+		keys: &'a TestKeysInterface,
+	) -> TestDeferredChainMonitor<'a> {
+		ChainMonitor::new(
+			Some(chain_source),
+			broadcaster,
+			logger,
+			fee_est,
+			persister,
+			keys,
+			keys.get_peer_storage_key(),
+			true,
+		)
+	}
+
+	/// Tests queueing and flushing of both `watch_channel` and `update_channel` operations
+	/// when `ChainMonitor` is in deferred mode, verifying that operations flow through to
+	/// `Persist` and that `channel_monitor_updated` is called on `Completed` status.
+	#[test]
+	fn test_queue_and_flush() {
+		let broadcaster = TestBroadcaster::new(Network::Testnet);
+		let fee_est = TestFeeEstimator::new(253);
+		let logger = TestLogger::new();
+		let persister = TestPersister::new();
+		let chain_source = TestChainSource::new(Network::Testnet);
+		let keys = TestKeysInterface::new(&[0; 32], Network::Testnet);
+		let deferred = create_deferred_chain_monitor(
+			&chain_source,
+			&broadcaster,
+			&logger,
+			&fee_est,
+			&persister,
+			&keys,
+		);
+
+		// Queue starts empty.
+		assert_eq!(deferred.pending_operation_count(), 0);
+
+		// Queue a watch_channel, verifying InProgress status.
+		let chan = ChannelId::from_bytes([1u8; 32]);
+		let monitor = crate::chain::channelmonitor::dummy_monitor(chan, |keys| {
+			TestChannelSigner::new(DynSigner::new(keys))
+		});
+		let status = Watch::watch_channel(&deferred, chan, monitor);
+		assert_eq!(status, Ok(ChannelMonitorUpdateStatus::InProgress));
+		assert_eq!(deferred.pending_operation_count(), 1);
+
+		// Nothing persisted yet — operations are only queued.
+		assert!(persister.new_channel_persistences.lock().unwrap().is_empty());
+
+		// Queue two updates after the watch. Update IDs must be sequential (starting
+		// from 1 since the initial monitor has update_id 0).
+		assert_eq!(
+			Watch::update_channel(&deferred, chan, &dummy_update(1, chan)),
+			ChannelMonitorUpdateStatus::InProgress
+		);
+		assert_eq!(
+			Watch::update_channel(&deferred, chan, &dummy_update(2, chan)),
+			ChannelMonitorUpdateStatus::InProgress
+		);
+		assert_eq!(deferred.pending_operation_count(), 3);
+
+		// Flush 2 of 3: persist_new_channel returns Completed (triggers
+		// channel_monitor_updated), update_persisted_channel returns InProgress (does not).
+		persister.set_update_ret(ChannelMonitorUpdateStatus::Completed);
+		persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
+		deferred.flush(2, &&logger);
+
+		assert_eq!(deferred.pending_operation_count(), 1);
+
+		// persist_new_channel was called for the watch.
+		assert_eq!(persister.new_channel_persistences.lock().unwrap().len(), 1);
+
+		// Because persist_new_channel returned Completed, channel_monitor_updated was called,
+		// so update_id 0 should no longer be pending.
+		let pending = deferred.list_pending_monitor_updates();
+		#[cfg(not(c_bindings))]
+		let pending_for_chan = pending.get(&chan).unwrap();
+		#[cfg(c_bindings)]
+		let pending_for_chan = &pending.iter().find(|(chan_id, _)| *chan_id == chan).unwrap().1;
+		assert!(!pending_for_chan.contains(&0));
+
+		// update_persisted_channel was called for update_id 1, and because it returned
+		// InProgress, update_id 1 remains pending.
+		let monitor_name = deferred.get_monitor(chan).unwrap().persistence_key();
+		assert!(persister
+			.offchain_monitor_updates
+			.lock()
+			.unwrap()
+			.get(&monitor_name)
+			.unwrap()
+			.contains(&1));
+		assert!(pending_for_chan.contains(&1));
+
+		// Flush remaining: update_persisted_channel returns Completed (default), triggers
+		// channel_monitor_updated.
+		deferred.flush(1, &&logger);
+		assert_eq!(deferred.pending_operation_count(), 0);
+
+		// update_persisted_channel was called for update_id 2.
+		assert!(persister
+			.offchain_monitor_updates
+			.lock()
+			.unwrap()
+			.get(&monitor_name)
+			.unwrap()
+			.contains(&2));
+
+		// update_id 1 is still pending from the InProgress earlier, but update_id 2 was
+		// completed in this flush so it is no longer pending.
+		let pending = deferred.list_pending_monitor_updates();
+		#[cfg(not(c_bindings))]
+		let pending_for_chan = pending.get(&chan).unwrap();
+		#[cfg(c_bindings)]
+		let pending_for_chan = &pending.iter().find(|(chan_id, _)| *chan_id == chan).unwrap().1;
+		assert!(pending_for_chan.contains(&1));
+		assert!(!pending_for_chan.contains(&2));
+
+		// Flushing an empty queue is a no-op.
+		let persist_count_before = persister.new_channel_persistences.lock().unwrap().len();
+		deferred.flush(0, &&logger);
+		assert_eq!(persister.new_channel_persistences.lock().unwrap().len(), persist_count_before);
+	}
+
+	/// Tests that `ChainMonitor` in deferred mode properly defers `watch_channel` and
+	/// `update_channel` operations, verifying correctness through a complete channel open
+	/// and payment flow. Operations are auto-flushed via the `TestChainMonitor`
+	/// `release_pending_monitor_events` helper.
+	#[test]
+	fn test_deferred_monitor_payment() {
+		let chanmon_cfgs = create_chanmon_cfgs(2);
+		let node_cfgs = create_node_cfgs_deferred(2, &chanmon_cfgs);
+		let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+		let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+		let chain_monitor_a = &nodes[0].chain_monitor.chain_monitor;
+		let chain_monitor_b = &nodes[1].chain_monitor.chain_monitor;
+
+		create_announced_chan_between_nodes(&nodes, 0, 1);
+
+		let (preimage, _hash, ..) = route_payment(&nodes[0], &[&nodes[1]], 10_000);
+		claim_payment(&nodes[0], &[&nodes[1]], preimage);
+
+		assert_eq!(chain_monitor_a.list_monitors().len(), 1);
+		assert_eq!(chain_monitor_b.list_monitors().len(), 1);
+		assert_eq!(chain_monitor_a.pending_operation_count(), 0);
+		assert_eq!(chain_monitor_b.pending_operation_count(), 0);
+	}
 }
diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs
index 641842d..596b242 100644
--- a/lightning/src/ln/functional_test_utils.rs
+++ b/lightning/src/ln/functional_test_utils.rs
@@ -4566,6 +4566,7 @@ pub fn create_chanmon_cfgs_internal(
 fn create_node_cfgs_internal<'a, F>(
 	node_count: usize, chanmon_cfgs: &'a Vec<TestChanMonCfg>,
 	persisters: Vec<&'a impl test_utils::SyncPersist>, message_router_constructor: F,
+	deferred: bool,
 ) -> Vec<NodeCfg<'a>>
 where
 	F: Fn(
@@ -4578,14 +4579,25 @@ where
 	for i in 0..node_count {
 		let cfg = &chanmon_cfgs[i];
 		let network_graph = Arc::new(NetworkGraph::new(Network::Testnet, &cfg.logger));
-		let chain_monitor = test_utils::TestChainMonitor::new(
-			Some(&cfg.chain_source),
-			&cfg.tx_broadcaster,
-			&cfg.logger,
-			&cfg.fee_estimator,
-			persisters[i],
-			&cfg.keys_manager,
-		);
+		let chain_monitor = if deferred {
+			test_utils::TestChainMonitor::new_deferred(
+				Some(&cfg.chain_source),
+				&cfg.tx_broadcaster,
+				&cfg.logger,
+				&cfg.fee_estimator,
+				persisters[i],
+				&cfg.keys_manager,
+			)
+		} else {
+			test_utils::TestChainMonitor::new(
+				Some(&cfg.chain_source),
+				&cfg.tx_broadcaster,
+				&cfg.logger,
+				&cfg.fee_estimator,
+				persisters[i],
+				&cfg.keys_manager,
+			)
+		};
 
 		let seed = [i as u8; 32];
 		nodes.push(NodeCfg {
@@ -4622,6 +4634,20 @@ pub fn create_node_cfgs<'a>(
 		chanmon_cfgs,
 		persisters,
 		test_utils::TestMessageRouter::new_default,
+		false,
+	)
+}
+
+pub fn create_node_cfgs_deferred<'a>(
+	node_count: usize, chanmon_cfgs: &'a Vec<TestChanMonCfg>,
+) -> Vec<NodeCfg<'a>> {
+	let persisters = chanmon_cfgs.iter().map(|c| &c.persister).collect();
+	create_node_cfgs_internal(
+		node_count,
+		chanmon_cfgs,
+		persisters,
+		test_utils::TestMessageRouter::new_default,
+		true,
 	)
 }
 
@@ -4634,6 +4660,7 @@ pub fn create_node_cfgs_with_persisters<'a>(
 		chanmon_cfgs,
 		persisters,
 		test_utils::TestMessageRouter::new_default,
+		false,
 	)
 }
 
@@ -4646,6 +4673,7 @@ pub fn create_node_cfgs_with_node_id_message_router<'a>(
 		chanmon_cfgs,
 		persisters,
 		test_utils::TestMessageRouter::new_node_id_router,
+		false,
 	)
 }
 
diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs
index 1009d2a..3541c82 100644
--- a/lightning/src/util/test_utils.rs
+++ b/lightning/src/util/test_utils.rs
@@ -508,6 +508,7 @@ pub struct TestChainMonitor<'a> {
 		&'a TestKeysInterface,
 	>,
 	pub keys_manager: &'a TestKeysInterface,
+	pub logger: &'a TestLogger,
 	/// If this is set to Some(), the next update_channel call (not watch_channel) must be a
 	/// ChannelForceClosed event for the given channel_id with should_broadcast set to the given
 	/// boolean.
@@ -523,6 +524,38 @@ impl<'a> TestChainMonitor<'a> {
 		chain_source: Option<&'a TestChainSource>, broadcaster: &'a dyn SyncBroadcaster,
 		logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator,
 		persister: &'a dyn SyncPersist, keys_manager: &'a TestKeysInterface,
+	) -> Self {
+		Self::with_deferred(
+			chain_source,
+			broadcaster,
+			logger,
+			fee_estimator,
+			persister,
+			keys_manager,
+			false,
+		)
+	}
+
+	pub fn new_deferred(
+		chain_source: Option<&'a TestChainSource>, broadcaster: &'a dyn SyncBroadcaster,
+		logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator,
+		persister: &'a dyn SyncPersist, keys_manager: &'a TestKeysInterface,
+	) -> Self {
+		Self::with_deferred(
+			chain_source,
+			broadcaster,
+			logger,
+			fee_estimator,
+			persister,
+			keys_manager,
+			true,
+		)
+	}
+
+	fn with_deferred(
+		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,
 	) -> Self {
 		Self {
 			added_monitors: Mutex::new(Vec::new()),
@@ -536,9 +569,10 @@ impl<'a> TestChainMonitor<'a> {
 				persister,
 				keys_manager,
 				keys_manager.get_peer_storage_key(),
-				false,
+				deferred,
 			),
 			keys_manager,
+			logger,
 			expect_channel_force_closed: Mutex::new(None),
 			expect_monitor_round_trip_fail: Mutex::new(None),
 			#[cfg(feature = "std")]
@@ -546,6 +580,10 @@ impl<'a> TestChainMonitor<'a> {
 		}
 	}
 
+	pub fn pending_operation_count(&self) -> usize {
+		self.chain_monitor.pending_operation_count()
+	}
+
 	pub fn complete_sole_pending_chan_update(&self, channel_id: &ChannelId) {
 		let (_, latest_update) =
 			self.latest_monitor_update_id.lock().unwrap().get(channel_id).unwrap().clone();
@@ -676,6 +714,12 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 	fn release_pending_monitor_events(
 		&self,
 	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
+		// Auto-flush pending operations so that the ChannelManager can pick up monitor
+		// completion events. When not in deferred mode the queue is empty so this only
+		// costs a lock acquisition. It ensures standard test helpers (route_payment, etc.)
+		// work with deferred chain monitors.
+		let count = self.chain_monitor.pending_operation_count();
+		self.chain_monitor.flush(count, &self.logger);
 		return self.chain_monitor.release_pending_monitor_events();
 	}
 }
@@ -835,6 +879,8 @@ pub struct TestPersister {
 	/// The queue of update statuses we'll return. If none are queued, ::Completed will always be
 	/// returned.
 	pub update_rets: Mutex<VecDeque<chain::ChannelMonitorUpdateStatus>>,
+	/// When we get a persist_new_channel call, we push the monitor name here.
+	pub new_channel_persistences: Mutex<Vec<MonitorName>>,
 	/// When we get an update_persisted_channel call *with* a ChannelMonitorUpdate, we insert the
 	/// [`ChannelMonitor::get_latest_update_id`] here.
 	pub offchain_monitor_updates: Mutex<HashMap<MonitorName, HashSet<u64>>>,
@@ -845,9 +891,15 @@ pub struct TestPersister {
 impl TestPersister {
 	pub fn new() -> Self {
 		let update_rets = Mutex::new(VecDeque::new());
+		let new_channel_persistences = Mutex::new(Vec::new());
 		let offchain_monitor_updates = Mutex::new(new_hash_map());
 		let chain_sync_monitor_persistences = Mutex::new(VecDeque::new());
-		Self { update_rets, offchain_monitor_updates, chain_sync_monitor_persistences }
+		Self {
+			update_rets,
+			new_channel_persistences,
+			offchain_monitor_updates,
+			chain_sync_monitor_persistences,
+		}
 	}
 
 	/// Queue an update status to return.
@@ -857,8 +909,9 @@ impl TestPersister {
 }
 impl<Signer: sign::ecdsa::EcdsaChannelSigner> Persist<Signer> for TestPersister {
 	fn persist_new_channel(
-		&self, _monitor_name: MonitorName, _data: &ChannelMonitor<Signer>,
+		&self, monitor_name: MonitorName, _data: &ChannelMonitor<Signer>,
 	) -> chain::ChannelMonitorUpdateStatus {
+		self.new_channel_persistences.lock().unwrap().push(monitor_name);
 		if let Some(update_ret) = self.update_rets.lock().unwrap().pop_front() {
 			return update_ret;
 		}