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.

Key changes:
- Add PendingWrite enum to represent queued write/remove operations
- Add pending_writes queue to MonitorUpdatingPersisterAsyncInner
- Add flush() 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

@@ -1349,6 +1349,11 @@ where
 			res?;
 		}
 
+		// Flush any pending monitor writes after channel manager persistence.
+		if let Err(e) = chain_monitor.flush() {
+			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 +1418,12 @@ where
 			channel_manager.get_cm().encode(),
 		)
 		.await?;
+
+	// Flush any pending monitor writes after final channel manager persistence.
+	if let Err(e) = chain_monitor.flush() {
+		log_error!(logger, "Failed to flush chain monitor: {}", e);
+	}
+
 	if let Some(ref scorer) = scorer {
 		kv_store
 			.write(
@@ -1731,6 +1742,11 @@ impl BackgroundProcessor {
 						channel_manager.get_cm().encode(),
 					))?;
 					log_trace!(logger, "Done persisting ChannelManager.");
+
+					// Flush any pending monitor writes after channel manager persistence.
+					if let Err(e) = chain_monitor.flush() {
+						log_error!(logger, "Failed to flush chain monitor: {}", e);
+					}
 				}
 
 				if let Some(liquidity_manager) = liquidity_manager.as_ref() {
@@ -1853,6 +1869,12 @@ impl BackgroundProcessor {
 				CHANNEL_MANAGER_PERSISTENCE_KEY,
 				channel_manager.get_cm().encode(),
 			)?;
+
+			// Flush any pending monitor writes after final channel manager persistence.
+			if let Err(e) = chain_monitor.flush() {
+				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,15 @@ pub trait Persist<ChannelSigner: EcdsaChannelSigner> {
 	fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> {
 		Vec::new()
 	}
+
+	/// Flushes any pending writes to the underlying storage.
+	///
+	/// For implementations that queue writes (returning [`ChannelMonitorUpdateStatus::InProgress`]
+	/// from persist methods), this method should write all 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) -> Result<Vec<(ChannelId, u64)>, io::Error>;
 }
 
 struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> {
@@ -272,7 +282,6 @@ pub struct AsyncPersister<
 	FE::Target: FeeEstimator,
 {
 	persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, BI, FE>,
-	event_notifier: Arc<Notifier>,
 }
 
 impl<
@@ -320,17 +329,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 +348,10 @@ where
 	fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> {
 		self.persister.get_and_clear_completed_updates()
 	}
+
+	fn flush(&self) -> Result<Vec<(ChannelId, u64)>, io::Error> {
+		crate::util::persist::poll_sync_future(self.persister.flush())
+	}
 }
 
 /// An implementation of [`chain::Watch`] for monitoring channels.
@@ -440,7 +451,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 +460,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 +752,22 @@ where
 			.collect()
 	}
 
+	/// Flushes any pending writes to the underlying storage.
+	///
+	/// For persisters that queue writes (returning [`ChannelMonitorUpdateStatus::InProgress`]
+	/// from persist methods), this method writes all 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) -> Result<Vec<(ChannelId, u64)>, io::Error> {
+		let completed = self.persister.flush()?;
+		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