Defer MonitorUpdatingPersister writes to flush()

Update MonitorUpdatingPersister and MonitorUpdatingPersisterAsync to
queue persist operations in memory instead of writing immediately to
disk. The Persist trait methods now return ChannelMonitorUpdateStatus::
InProgress and the actual writes happen when flush() is called.

This fixes a race condition that could cause channel force closures:
previously, if the node crashed after writing channel monitors but
before writing the channel manager, the monitors would be ahead of
the manager on restart. By deferring monitor writes until after the
channel manager is persisted (via flush()), we ensure the manager is
always at least as up-to-date as the monitors.

The flush() method takes an optional count parameter to flush only a
specific number of queued writes. The background processor captures
the queue size before persisting the channel manager, then flushes
exactly that many writes afterward. This prevents flushing monitor
updates that arrived after the manager state was captured.

Key changes:
- Add PendingWrite enum to represent queued write/remove operations
- Add pending_writes queue to MonitorUpdatingPersisterAsyncInner
- Add pending_write_count() and flush(count) to Persist trait and ChainMonitor
- ChainMonitor::flush() calls channel_monitor_updated for each completed write
- Update Persist impl to queue writes and return InProgress
- Call flush() in background processor after channel manager persistence
- Remove unused event_notifier from AsyncPersister

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

Author: joostjager

lightning-background-processor/src/lib.rs

@@ -1152,6 +1152,11 @@ where
 
 		let mut futures = Joiner::new();
 
+		// 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_write_count();
+
 		if channel_manager.get_cm().get_and_clear_needs_persistence() {
 			log_trace!(logger, "Persisting ChannelManager...");
 
@@ -1349,6 +1354,14 @@ where
 			res?;
 		}
 
+		// Flush the monitor writes that were pending before we persisted the channel manager.
+		// Any writes that arrived after are left in the queue for the next iteration.
+		if pending_monitor_writes > 0 {
+			if let Err(e) = chain_monitor.flush(Some(pending_monitor_writes)) {
+				log_error!(logger, "Failed to flush chain monitor: {}", e);
+			}
+		}
+
 		match check_and_reset_sleeper(&mut last_onion_message_handler_call, || {
 			sleeper(ONION_MESSAGE_HANDLER_TIMER)
 		}) {
@@ -1413,6 +1426,12 @@ where
 			channel_manager.get_cm().encode(),
 		)
 		.await?;
+
+	// Flush all pending monitor writes after final channel manager persistence.
+	if let Err(e) = chain_monitor.flush(None) {
+		log_error!(logger, "Failed to flush chain monitor: {}", e);
+	}
+
 	if let Some(ref scorer) = scorer {
 		kv_store
 			.write(
@@ -1722,6 +1741,9 @@ impl BackgroundProcessor {
 					channel_manager.get_cm().timer_tick_occurred();
 					last_freshness_call = Instant::now();
 				}
+				// Capture the number of pending monitor writes before persisting the channel manager.
+				let pending_monitor_writes = chain_monitor.pending_write_count();
+
 				if channel_manager.get_cm().get_and_clear_needs_persistence() {
 					log_trace!(logger, "Persisting ChannelManager...");
 					(kv_store.write(
@@ -1733,6 +1755,13 @@ impl BackgroundProcessor {
 					log_trace!(logger, "Done persisting ChannelManager.");
 				}
 
+				// Flush the monitor writes that were pending before we persisted the channel manager.
+				if pending_monitor_writes > 0 {
+					if let Err(e) = chain_monitor.flush(Some(pending_monitor_writes)) {
+						log_error!(logger, "Failed to flush chain monitor: {}", e);
+					}
+				}
+
 				if let Some(liquidity_manager) = liquidity_manager.as_ref() {
 					log_trace!(logger, "Persisting LiquidityManager...");
 					let _ = liquidity_manager.get_lm().persist().map_err(|e| {
@@ -1853,6 +1882,12 @@ impl BackgroundProcessor {
 				CHANNEL_MANAGER_PERSISTENCE_KEY,
 				channel_manager.get_cm().encode(),
 			)?;
+
+			// Flush all pending monitor writes after final channel manager persistence.
+			if let Err(e) = chain_monitor.flush(None) {
+				log_error!(logger, "Failed to flush chain monitor: {}", e);
+			}
+
 			if let Some(ref scorer) = scorer {
 				kv_store.write(
 					SCORER_PERSISTENCE_PRIMARY_NAMESPACE,

lightning/src/chain/chainmonitor.rs

@@ -39,6 +39,7 @@ use crate::chain::channelmonitor::{
 use crate::chain::transaction::{OutPoint, TransactionData};
 use crate::chain::{BestBlock, ChannelMonitorUpdateStatus, Filter, WatchedOutput};
 use crate::events::{self, Event, EventHandler, ReplayEvent};
+use crate::io;
 use crate::ln::channel_state::ChannelDetails;
 #[cfg(peer_storage)]
 use crate::ln::msgs::PeerStorage;
@@ -208,6 +209,26 @@ pub trait Persist<ChannelSigner: EcdsaChannelSigner> {
 	fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> {
 		Vec::new()
 	}
+
+	/// Returns the number of pending writes in the queue.
+	///
+	/// This can be used to capture the queue size before persisting the channel manager,
+	/// then pass that count to [`Self::flush`] to only flush those specific updates.
+	fn pending_write_count(&self) -> usize {
+		0
+	}
+
+	/// Flushes pending writes to the underlying storage.
+	///
+	/// If `count` is `Some(n)`, only the first `n` pending writes are flushed.
+	/// If `count` is `None`, all pending writes are flushed.
+	///
+	/// For implementations that queue writes (returning [`ChannelMonitorUpdateStatus::InProgress`]
+	/// from persist methods), this method should write queued data to storage.
+	///
+	/// Returns the list of completed updates (channel_id, update_id) on success, or an error if
+	/// any write failed.
+	fn flush(&self, count: Option<usize>) -> Result<Vec<(ChannelId, u64)>, io::Error>;
 }
 
 struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> {
@@ -272,7 +293,6 @@ pub struct AsyncPersister<
 	FE::Target: FeeEstimator,
 {
 	persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, BI, FE>,
-	event_notifier: Arc<Notifier>,
 }
 
 impl<
@@ -320,17 +340,15 @@ where
 		&self, monitor_name: MonitorName,
 		monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>,
 	) -> ChannelMonitorUpdateStatus {
-		let notifier = Arc::clone(&self.event_notifier);
-		self.persister.spawn_async_persist_new_channel(monitor_name, monitor, notifier);
+		self.persister.queue_new_channel(monitor_name, monitor);
 		ChannelMonitorUpdateStatus::InProgress
 	}
 
 	fn update_persisted_channel(
 		&self, monitor_name: MonitorName, monitor_update: Option<&ChannelMonitorUpdate>,
 		monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>,
 	) -> ChannelMonitorUpdateStatus {
-		let notifier = Arc::clone(&self.event_notifier);
-		self.persister.spawn_async_update_channel(monitor_name, monitor_update, monitor, notifier);
+		self.persister.queue_channel_update(monitor_name, monitor_update, monitor);
 		ChannelMonitorUpdateStatus::InProgress
 	}
 
@@ -341,6 +359,14 @@ where
 	fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> {
 		self.persister.get_and_clear_completed_updates()
 	}
+
+	fn pending_write_count(&self) -> usize {
+		self.persister.pending_write_count()
+	}
+
+	fn flush(&self, count: Option<usize>) -> Result<Vec<(ChannelId, u64)>, io::Error> {
+		crate::util::persist::poll_sync_future(self.persister.flush(count))
+	}
 }
 
 /// An implementation of [`chain::Watch`] for monitoring channels.
@@ -440,7 +466,6 @@ impl<
 		persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, T, F>, _entropy_source: ES,
 		_our_peerstorage_encryption_key: PeerStorageKey,
 	) -> Self {
-		let event_notifier = Arc::new(Notifier::new());
 		Self {
 			monitors: RwLock::new(new_hash_map()),
 			chain_source,
@@ -450,8 +475,8 @@ impl<
 			_entropy_source,
 			pending_monitor_events: Mutex::new(Vec::new()),
 			highest_chain_height: AtomicUsize::new(0),
-			event_notifier: Arc::clone(&event_notifier),
-			persister: AsyncPersister { persister, event_notifier },
+			event_notifier: Arc::new(Notifier::new()),
+			persister: AsyncPersister { persister },
 			pending_send_only_events: Mutex::new(Vec::new()),
 			#[cfg(peer_storage)]
 			our_peerstorage_encryption_key: _our_peerstorage_encryption_key,
@@ -742,6 +767,33 @@ where
 			.collect()
 	}
 
+	/// Returns the number of pending writes in the persister queue.
+	///
+	/// This can be used to capture the queue size before persisting the channel manager,
+	/// then pass that count to [`Self::flush`] to only flush those specific updates.
+	pub fn pending_write_count(&self) -> usize {
+		self.persister.pending_write_count()
+	}
+
+	/// Flushes pending writes to the underlying storage.
+	///
+	/// If `count` is `Some(n)`, only the first `n` pending writes are flushed.
+	/// If `count` is `None`, all pending writes are flushed.
+	///
+	/// For persisters that queue writes (returning [`ChannelMonitorUpdateStatus::InProgress`]
+	/// from persist methods), this method writes queued data to storage and signals
+	/// completion to the channel manager via [`Self::channel_monitor_updated`].
+	///
+	/// Returns the list of completed updates (channel_id, update_id) on success, or an error if
+	/// any write failed. Note that even if an error is returned, some writes may have succeeded.
+	pub fn flush(&self, count: Option<usize>) -> Result<Vec<(ChannelId, u64)>, io::Error> {
+		let completed = self.persister.flush(count)?;
+		for (channel_id, update_id) in &completed {
+			let _ = self.channel_monitor_updated(*channel_id, *update_id);
+		}
+		Ok(completed)
+	}
+
 	#[cfg(any(test, feature = "_test_utils"))]
 	pub fn remove_monitor(&self, channel_id: &ChannelId) -> ChannelMonitor<ChannelSigner> {
 		self.monitors.write().unwrap().remove(channel_id).unwrap().monitor