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 a count parameter specifying how many queued
writes to flush. 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 with FullMonitor and Update variants for queued writes
- 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
- Stale update cleanup happens in flush() after full monitor is written
- Call flush() in background processor after channel manager persistence

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,15 @@ 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 {
+			match chain_monitor.flush(pending_monitor_writes) {
+				Ok(()) => log_trace!(logger, "Flushed {} monitor writes", pending_monitor_writes),
+				Err(e) => 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 +1427,16 @@ where
 			channel_manager.get_cm().encode(),
 		)
 		.await?;
+
+	// Flush all pending monitor writes after final channel manager persistence.
+	let pending_monitor_writes = chain_monitor.pending_write_count();
+	if pending_monitor_writes > 0 {
+		match chain_monitor.flush(pending_monitor_writes) {
+			Ok(()) => log_trace!(logger, "Flushed {} monitor writes", pending_monitor_writes),
+			Err(e) => log_error!(logger, "Failed to flush chain monitor: {}", e),
+		}
+	}
+
 	if let Some(ref scorer) = scorer {
 		kv_store
 			.write(
@@ -1722,6 +1746,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 +1760,16 @@ 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 {
+					match chain_monitor.flush(pending_monitor_writes) {
+						Ok(()) => {
+							log_trace!(logger, "Flushed {} monitor writes", pending_monitor_writes)
+						},
+						Err(e) => 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 +1890,18 @@ impl BackgroundProcessor {
 				CHANNEL_MANAGER_PERSISTENCE_KEY,
 				channel_manager.get_cm().encode(),
 			)?;
+
+			// Flush all pending monitor writes after final channel manager persistence.
+			let pending_monitor_writes = chain_monitor.pending_write_count();
+			if pending_monitor_writes > 0 {
+				match chain_monitor.flush(pending_monitor_writes) {
+					Ok(()) => {
+						log_trace!(logger, "Flushed {} monitor writes", pending_monitor_writes)
+					},
+					Err(e) => 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;
@@ -198,16 +199,23 @@ pub trait Persist<ChannelSigner: EcdsaChannelSigner> {
 	/// the monitor already exists in the archive.
 	fn archive_persisted_channel(&self, monitor_name: MonitorName);
 
-	/// Fetches the set of [`ChannelMonitorUpdate`]s, previously persisted with
-	/// [`Self::update_persisted_channel`], which have completed.
+	/// Returns the number of pending writes in the queue.
 	///
-	/// Returning an update here is equivalent to calling
-	/// [`ChainMonitor::channel_monitor_updated`]. Because of this, this method is defaulted and
-	/// hidden in the docs.
-	#[doc(hidden)]
-	fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> {
-		Vec::new()
+	/// 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.
+	///
+	/// The `count` parameter specifies how many pending writes to flush.
+	///
+	/// For implementations that queue writes (returning [`ChannelMonitorUpdateStatus::InProgress`]
+	/// from persist methods), this method should write queued data to storage.
+	///
+	/// Returns the list of completed monitor updates (channel_id, update_id) that were flushed.
+	fn flush(&self, count: usize) -> Result<Vec<(ChannelId, u64)>, io::Error>;
 }
 
 struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> {
@@ -272,7 +280,6 @@ pub struct AsyncPersister<
 	FE::Target: FeeEstimator,
 {
 	persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, BI, FE>,
-	event_notifier: Arc<Notifier>,
 }
 
 impl<
@@ -320,26 +327,28 @@ 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
 	}
 
 	fn archive_persisted_channel(&self, monitor_name: MonitorName) {
 		self.persister.spawn_async_archive_persisted_channel(monitor_name);
 	}
 
-	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: usize) -> Result<Vec<(ChannelId, u64)>, io::Error> {
+		crate::util::persist::poll_sync_future(self.persister.flush(count))
 	}
 }
 
@@ -440,7 +449,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 +458,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 +750,30 @@ 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`].
+	pub fn flush(&self, count: usize) -> Result<(), 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(())
+	}
+
 	#[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
@@ -1497,9 +1529,6 @@ where
 	fn release_pending_monitor_events(
 		&self,
 	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
-		for (channel_id, update_id) in self.persister.get_and_clear_completed_updates() {
-			let _ = self.channel_monitor_updated(channel_id, update_id);
-		}
 		let mut pending_monitor_events = self.pending_monitor_events.lock().unwrap().split_off(0);
 		for monitor_state in self.monitors.read().unwrap().values() {
 			let monitor_events = monitor_state.monitor.get_and_clear_pending_monitor_events();