Merge PR 'Persistent `MonitorEvent`s' (#4491)

### fuzz/src/chanmon_consistency.rs
@@ -1354,13 +1354,15 @@ impl<'a> HarnessNode<'a> {
 					&self.broadcaster,
 					&self.fee_estimator,
 					&self.logger,
+					&self.keys_manager,
 				);
 				monitor.best_block_updated(
 					header,
 					height,
 					&self.broadcaster,
 					&self.fee_estimator,
 					&self.logger,
+					&self.keys_manager,
 				);
 			}
 			let (header, txn) = chain_state.block_at(target_height);
@@ -1373,6 +1375,7 @@ impl<'a> HarnessNode<'a> {
 					&self.broadcaster,
 					&self.fee_estimator,
 					&self.logger,
+					&self.keys_manager,
 				);
 			}
 		}

### lightning-block-sync/src/init.rs
@@ -117,7 +117,8 @@ where
 /// 	};
 ///
 /// 	// Synchronize any channel monitors and the channel manager to be on the best block.
-/// 	let mut monitor_listener = (monitor, &*tx_broadcaster, &*fee_estimator, &*logger);
+/// 	let mut monitor_listener =
+/// 		(monitor, &*tx_broadcaster, &*fee_estimator, &*logger, &*entropy_source);
 /// 	let listeners = vec![
 /// 		(monitor_best_block, &monitor_listener as &dyn chain::Listen),
 /// 		(manager_best_block, &manager as &dyn chain::Listen),

### lightning/src/chain/chainmonitor.rs
@@ -33,8 +33,8 @@ use crate::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 #[cfg(peer_storage)]
 use crate::chain::channelmonitor::write_chanmon_internal;
 use crate::chain::channelmonitor::{
-	Balance, ChannelMonitor, ChannelMonitorUpdate, MonitorEvent, TransactionOutputs,
-	WithChannelMonitor,
+	random_monitor_event_id, Balance, ChannelMonitor, ChannelMonitorUpdate, MonitorEvent,
+	TransactionOutputs, WithChannelMonitor,
 };
 use crate::chain::transaction::{OutPoint, TransactionData};
 use crate::chain::{BlockLocator, ChannelMonitorUpdateStatus, WatchedOutput};
@@ -66,6 +66,21 @@ use core::iter::Cycle;
 use core::ops::Deref;
 use core::sync::atomic::{AtomicUsize, Ordering};
 
+/// Identifies the source of a [`MonitorEvent`] for acknowledgment via
+/// [`chain::Watch::ack_monitor_event`] once the event has been processed.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct MonitorEventSource {
+	/// The randomly-generated event ID.
+	pub event_id: u128,
+	/// The channel from which the [`MonitorEvent`] originated.
+	pub channel_id: ChannelId,
+}
+
+impl_ser_tlv_based!(MonitorEventSource, {
+	(1, event_id, required),
+	(3, channel_id, required),
+});
+
 /// A pending operation queued for later execution when `ChainMonitor` is in deferred mode.
 enum PendingMonitorOp<ChannelSigner: EcdsaChannelSigner> {
 	/// A new monitor to insert and persist.
@@ -365,10 +380,10 @@ pub struct ChainMonitor<
 	logger: L,
 	fee_estimator: F,
 	persister: P,
-	_entropy_source: ES,
+	entropy_source: ES,
 	/// "User-provided" (ie persistence-completion/-failed) [`MonitorEvent`]s. These came directly
 	/// from the user and not from a [`ChannelMonitor`].
-	pending_monitor_events: Mutex<Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)>>,
+	pending_monitor_events: Mutex<Vec<(OutPoint, ChannelId, Vec<(u128, MonitorEvent)>, PublicKey)>>,
 	/// The best block height seen, used as a proxy for the passage of time.
 	highest_chain_height: AtomicUsize,
 
@@ -425,7 +440,7 @@ where
 	/// 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,
-		persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, T, F>, _entropy_source: ES,
+		persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, T, F>, entropy_source: ES,
 		_our_peerstorage_encryption_key: PeerStorageKey, deferred: bool,
 	) -> Self {
 		let event_notifier = Arc::new(Notifier::new());
@@ -435,7 +450,7 @@ where
 			broadcaster,
 			logger,
 			fee_estimator: feeest,
-			_entropy_source,
+			entropy_source,
 			pending_monitor_events: Mutex::new(Vec::new()),
 			highest_chain_height: AtomicUsize::new(0),
 			event_notifier: Arc::clone(&event_notifier),
@@ -647,7 +662,7 @@ where
 	/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
 	pub fn new(
 		chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, persister: P,
-		_entropy_source: ES, _our_peerstorage_encryption_key: PeerStorageKey, deferred: bool,
+		entropy_source: ES, _our_peerstorage_encryption_key: PeerStorageKey, deferred: bool,
 	) -> Self {
 		Self {
 			monitors: RwLock::new(new_hash_map()),
@@ -656,7 +671,7 @@ where
 			logger,
 			fee_estimator: feeest,
 			persister,
-			_entropy_source,
+			entropy_source,
 			pending_monitor_events: Mutex::new(Vec::new()),
 			highest_chain_height: AtomicUsize::new(0),
 			event_notifier: Arc::new(Notifier::new()),
@@ -748,6 +763,26 @@ where
 		self.monitors.write().unwrap().remove(channel_id).unwrap().monitor
 	}
 
+	/// Pushes a [`MonitorEvent::Completed`] to be provided to the [`ChannelManager`] in the next
+	/// [`chain::Watch::release_pending_monitor_events`] call.
+	///
+	/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
+	fn push_update_completed_event(
+		&self, funding_txo: OutPoint, channel_id: ChannelId, monitor_update_id: u64,
+		counterparty_node_id: PublicKey,
+	) {
+		let event_id = random_monitor_event_id(&self.entropy_source);
+		self.pending_monitor_events.lock().unwrap().push((
+			funding_txo,
+			channel_id,
+			vec![(
+				event_id,
+				MonitorEvent::Completed { funding_txo, channel_id, monitor_update_id },
+			)],
+			counterparty_node_id,
+		));
+	}
+
 	/// Indicates the persistence of a [`ChannelMonitor`] has completed after
 	/// [`ChannelMonitorUpdateStatus::InProgress`] was returned from an update operation.
 	///
@@ -801,17 +836,12 @@ where
 			// Completed event.
 			return Ok(());
 		}
-		let funding_txo = monitor_data.monitor.get_funding_txo();
-		self.pending_monitor_events.lock().unwrap().push((
-			funding_txo,
+		self.push_update_completed_event(
+			monitor_data.monitor.get_funding_txo(),
 			channel_id,
-			vec![MonitorEvent::Completed {
-				funding_txo,
-				channel_id,
-				monitor_update_id: monitor_data.monitor.get_latest_update_id(),
-			}],
+			monitor_data.monitor.get_latest_update_id(),
 			monitor_data.monitor.get_counterparty_node_id(),
-		));
+		);
 
 		self.event_notifier.notify();
 		Ok(())
@@ -824,14 +854,12 @@ where
 	pub fn force_channel_monitor_updated(&self, channel_id: ChannelId, monitor_update_id: u64) {
 		let monitors = self.monitors.read().unwrap();
 		let monitor = &monitors.get(&channel_id).unwrap().monitor;
-		let counterparty_node_id = monitor.get_counterparty_node_id();
-		let funding_txo = monitor.get_funding_txo();
-		self.pending_monitor_events.lock().unwrap().push((
-			funding_txo,
+		self.push_update_completed_event(
+			monitor.get_funding_txo(),
 			channel_id,
-			vec![MonitorEvent::Completed { funding_txo, channel_id, monitor_update_id }],
-			counterparty_node_id,
-		));
+			monitor_update_id,
+			monitor.get_counterparty_node_id(),
+		);
 		self.event_notifier.notify();
 	}
 
@@ -983,7 +1011,7 @@ where
 	#[cfg(peer_storage)]
 	fn send_peer_storage(&self, their_node_id: PublicKey) {
 		let mut monitors_list: Vec<PeerStorageMonitorHolder> = Vec::new();
-		let random_bytes = self._entropy_source.get_secure_random_bytes();
+		let random_bytes = self.entropy_source.get_secure_random_bytes();
 
 		const MAX_PEER_STORAGE_SIZE: usize = 65531;
 		const USIZE_LEN: usize = core::mem::size_of::<usize>();
@@ -1182,6 +1210,7 @@ where
 					&self.broadcaster,
 					&self.fee_estimator,
 					&self.logger,
+					&self.entropy_source,
 				);
 
 				let update_id = update.update_id;
@@ -1269,18 +1298,12 @@ where
 					// Push a Completed event into pending_monitor_events so it gets
 					// picked up after the per-monitor events in the next
 					// release_pending_monitor_events call.
-					let funding_txo = monitor.get_funding_txo();
-					let channel_id = monitor.channel_id();
-					self.pending_monitor_events.lock().unwrap().push((
-						funding_txo,
-						channel_id,
-						vec![MonitorEvent::Completed {
-							funding_txo,
-							channel_id,
-							monitor_update_id: monitor.get_latest_update_id(),
-						}],
+					self.push_update_completed_event(
+						monitor.get_funding_txo(),
+						monitor.channel_id(),
+						monitor.get_latest_update_id(),
 						monitor.get_counterparty_node_id(),
-					));
+					);
 					log_debug!(
 						logger,
 						"Deferring completion of ChannelMonitorUpdate id {:?} (channel is post-close)",
@@ -1460,6 +1483,7 @@ where
 				&self.broadcaster,
 				&self.fee_estimator,
 				&self.logger,
+				&self.entropy_source,
 			)
 		});
 
@@ -1487,6 +1511,7 @@ where
 				&self.broadcaster,
 				&self.fee_estimator,
 				&self.logger,
+				&self.entropy_source,
 			);
 		}
 	}
@@ -1520,6 +1545,7 @@ where
 				&self.broadcaster,
 				&self.fee_estimator,
 				&self.logger,
+				&self.entropy_source,
 			)
 		});
 		// Assume we may have some new events and wake the event processor
@@ -1535,6 +1561,7 @@ where
 				&self.broadcaster,
 				&self.fee_estimator,
 				&self.logger,
+				&self.entropy_source,
 			);
 		}
 	}
@@ -1556,6 +1583,7 @@ where
 				&self.broadcaster,
 				&self.fee_estimator,
 				&self.logger,
+				&self.entropy_source,
 			)
 		});
 
@@ -1645,7 +1673,7 @@ where
 
 	fn release_pending_monitor_events(
 		&self,
-	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
+	) -> Vec<(OutPoint, ChannelId, Vec<(u128, MonitorEvent)>, PublicKey)> {
 		for (channel_id, update_id) in self.persister.get_and_clear_completed_updates() {
 			let _ = self.channel_monitor_updated(channel_id, update_id);
 		}
@@ -1671,6 +1699,18 @@ where
 		pending_monitor_events.extend(self.pending_monitor_events.lock().unwrap().split_off(0));
 		pending_monitor_events
 	}
+
+	fn ack_monitor_event(&self, source: MonitorEventSource) {
+		let monitors = self.monitors.read().unwrap();
+		if let Some(monitor_state) = monitors.get(&source.channel_id) {
+			monitor_state.monitor.ack_monitor_event(source.event_id);
+		} else {
+			// A monitor is only archived once all of its events have been acknowledged, but an
+			// acknowledgement may be replayed after the monitor was archived (e.g. if the
+			// `ChannelManager` was last persisted before it processed the event that triggered
+			// the original acknowledgement), so simply ignore acks for missing monitors.
+		}
+	}
 }
 
 impl<

### lightning/src/chain/channelmonitor.rs
@@ -69,8 +69,8 @@ use crate::util::byte_utils;
 use crate::util::logger::{Logger, WithContext};
 use crate::util::persist::MonitorName;
 use crate::util::ser::{
-	MaybeReadable, Readable, ReadableArgs, RequiredWrapper, UpgradableRequired, Writeable, Writer,
-	U48,
+	Iterable, MaybeReadable, Readable, ReadableArgs, RequiredWrapper, UpgradableRequired,
+	Writeable, Writer, U48,
 };
 
 #[allow(unused_imports)]
@@ -184,8 +184,23 @@ impl Readable for ChannelMonitorUpdate {
 	}
 }
 
-/// An event to be processed by the ChannelManager.
-#[derive(Clone, PartialEq, Eq)]
+/// Generates a random ID used to identify a [`MonitorEvent`] until it is acknowledged.
+pub(super) fn random_monitor_event_id<ES: EntropySource>(entropy_source: ES) -> u128 {
+	let mut random_bytes = [0u8; 16];
+	random_bytes.copy_from_slice(&entropy_source.get_secure_random_bytes()[..16]);
+	u128::from_be_bytes(random_bytes)
+}
+
+fn push_monitor_event<ES: EntropySource>(
+	pending_monitor_events: &mut Vec<(u128, MonitorEvent)>, event: MonitorEvent, entropy_source: ES,
+) {
+	let id = random_monitor_event_id(entropy_source);
+	pending_monitor_events.push((id, event));
+}
+
+/// An event to be processed by the ChannelManager. Will be re-provided to the ChannelManager on
+/// startup until persistently acked via [`chain::Watch::ack_monitor_event`].
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub enum MonitorEvent {
 	/// A monitor event containing an HTLCUpdate.
 	HTLCEvent(HTLCUpdate),
@@ -249,7 +264,7 @@ impl_writeable_tlv_based_enum_upgradable_legacy!(MonitorEvent,
 /// Simple structure sent back by `chain::Watch` when an HTLC from a forward channel is detected on
 /// chain. Used to update the corresponding HTLC in the backward channel. Failing to pass the
 /// preimage claim backward will lead to loss of funds.
-#[derive(Clone, PartialEq, Eq)]
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub struct HTLCUpdate {
 	pub(crate) payment_hash: PaymentHash,
 	pub(crate) payment_preimage: Option<PaymentPreimage>,
@@ -1295,7 +1310,13 @@ pub(crate) struct ChannelMonitorImpl<Signer: EcdsaChannelSigner> {
 	// Note that because the `event_lock` in `ChainMonitor` is only taken in
 	// block/transaction-connected events and *not* during block/transaction-disconnected events,
 	// we further MUST NOT generate events during block/transaction-disconnection.
-	pending_monitor_events: Vec<MonitorEvent>,
+	pending_monitor_events: Vec<(u128, MonitorEvent)>,
+	// `MonitorEvent`s that have been provided to the `ChannelManager` via
+	// [`ChannelMonitor::get_and_clear_pending_monitor_events`] and are awaiting
+	// [`ChannelMonitor::ack_monitor_event`] for removal. If an event in this queue is not acked, it
+	// will be re-provided to the `ChannelManager` on startup; this field is not persisted
+	// and any events here will move back to `pending_monitor_events` after a restart.
+	provided_monitor_events: Vec<(u128, MonitorEvent)>,
 
 	pub(super) pending_events: Vec<Event>,
 	pub(super) is_processing_pending_events: bool,
@@ -1690,7 +1711,7 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 		&(channel_monitor
 			.pending_monitor_events
 			.iter()
-			.filter(|ev| match ev {
+			.filter(|(_, ev)| match ev {
 				MonitorEvent::HTLCEvent(_) => true,
 				MonitorEvent::HolderForceClosed(_) => true,
 				MonitorEvent::HolderForceClosedWithInfo { .. } => true,
@@ -1699,7 +1720,7 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 			.count() as u64)
 			.to_be_bytes(),
 	)?;
-	for event in channel_monitor.pending_monitor_events.iter() {
+	for (_, event) in channel_monitor.pending_monitor_events.iter() {
 		match event {
 			MonitorEvent::HTLCEvent(upd) => {
 				0u8.write(writer)?;
@@ -1744,19 +1765,29 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 	channel_monitor.lockdown_from_offchain.write(writer)?;
 	channel_monitor.holder_tx_signed.write(writer)?;
 
-	// If we have a `HolderForceClosedWithInfo` event, we need to write the `HolderForceClosed` for backwards compatibility.
-	let pending_monitor_events =
-		match channel_monitor.pending_monitor_events.iter().find(|ev| match ev {
-			MonitorEvent::HolderForceClosedWithInfo { .. } => true,
-			_ => false,
-		}) {
-			Some(MonitorEvent::HolderForceClosedWithInfo { outpoint, .. }) => {
-				let mut pending_monitor_events = channel_monitor.pending_monitor_events.clone();
-				pending_monitor_events.push(MonitorEvent::HolderForceClosed(*outpoint));
-				pending_monitor_events
-			},
-			_ => channel_monitor.pending_monitor_events.clone(),
-		};
+	// If we have a `HolderForceClosedWithInfo` event, we need to write the `HolderForceClosed`
+	// for backwards compatibility.
+	let holder_force_closed_compat =
+		channel_monitor.pending_monitor_events.iter().find_map(|(_, ev)| {
+			if let MonitorEvent::HolderForceClosedWithInfo { outpoint, .. } = ev {
+				Some(MonitorEvent::HolderForceClosed(*outpoint))
+			} else {
+				None
+			}
+		});
+	let pending_monitor_events_legacy = Some(Iterable(
+		channel_monitor
+			.pending_monitor_events
+			.iter()
+			.map(|(_, ev)| ev)
+			.chain(holder_force_closed_compat.as_ref()),
+	));
+	let pending_mon_evs_with_ids = Some(Iterable(
+		channel_monitor
+			.provided_monitor_events
+			.iter()
+			.chain(channel_monitor.pending_monitor_events.iter()),
+	));
 
 	let legacy_alternative_funding_confirmed = channel_monitor
 		.alternative_funding_confirmed
@@ -1767,7 +1798,7 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 	write_tlv_fields!(writer, {
 		(1, channel_monitor.funding_spend_confirmed, option),
 		(3, channel_monitor.htlcs_resolved_on_chain, required_vec),
-		(5, pending_monitor_events, required_vec),
+		(5, pending_monitor_events_legacy, option), // Equivalent to optional_vec because Iterable also writes as WithoutLength
 		(7, channel_monitor.funding_spend_seen, required),
 		(9, channel_monitor.counterparty_node_id, required),
 		(11, channel_monitor.confirmed_commitment_tx_counterparty_output, option),
@@ -1790,6 +1821,7 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 		(41, channel_monitor.funding.contribution, option),
 		(43, channel_monitor.funding_tx_confirmed_in, option),
 		(45, alternative_funding_confirmed_block, option),
+		(47, pending_mon_evs_with_ids, option),
 	});
 
 	Ok(())
@@ -1975,6 +2007,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 
 			payment_preimages: new_hash_map(),
 			pending_monitor_events: Vec::new(),
+			provided_monitor_events: Vec::new(),
 			pending_events: Vec::new(),
 			is_processing_pending_events: false,
 
@@ -2112,12 +2145,18 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	/// itself.
 	///
 	/// panics if the given update is not the next update by update_id.
-	pub fn update_monitor<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	pub fn update_monitor<
+		B: BroadcasterInterface,
+		F: FeeEstimator,
+		L: Logger,
+		ES: EntropySource,
+	>(
 		&self, updates: &ChannelMonitorUpdate, broadcaster: &B, fee_estimator: &F, logger: &L,
+		entropy_source: &ES,
 	) -> Result<(), ()> {
 		let mut inner = self.inner.lock().unwrap();
 		let logger = WithChannelMonitor::from_impl(logger, &*inner, None);
-		inner.update_monitor(updates, broadcaster, fee_estimator, &logger)
+		inner.update_monitor(updates, broadcaster, fee_estimator, &logger, entropy_source)
 	}
 
 	/// Gets the update_id from the latest ChannelMonitorUpdate which was applied to this
@@ -2187,12 +2226,58 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		}
 	}
 
-	/// Get the list of HTLCs who's status has been updated on chain. This should be called by
-	/// ChannelManager via [`chain::Watch::release_pending_monitor_events`].
-	pub fn get_and_clear_pending_monitor_events(&self) -> Vec<MonitorEvent> {
+	/// Get the list of HTLCs whose status has been updated. This should be called by ChannelManager
+	/// via [`chain::Watch::release_pending_monitor_events`].
+	///
+	/// Returned events are retained internally until [Self::ack_monitor_event] is called with their
+	/// ID.
+	pub fn get_and_clear_pending_monitor_events(&self) -> Vec<(u128, MonitorEvent)> {
 		self.inner.lock().unwrap().get_and_clear_pending_monitor_events()
 	}
 
+	/// Removes a [`MonitorEvent`] by its event ID, acknowledging that it has been processed.
+	/// Generally called by [`chain::Watch::ack_monitor_event`].
+	pub fn ack_monitor_event(&self, event_id: u128) {
+		let inner = &mut *self.inner.lock().unwrap();
+		inner.ack_monitor_event(event_id);
+	}
+
+	/// Copies [`MonitorEvent`] state from `other` into `self`.
+	/// Used in tests to align transient runtime state before equality comparison after a
+	/// serialization round-trip, where `self` is the round-tripped monitor and `other` is the
+	/// original.
+	#[cfg(any(test, feature = "_test_utils"))]
+	pub fn copy_monitor_event_state(&self, other: &ChannelMonitor<Signer>) {
+		let (provided, pending) = {
+			let other_inner = other.inner.lock().unwrap();
+			(
+				other_inner.provided_monitor_events.clone(),
+				other_inner.pending_monitor_events.clone(),
+			)
+		};
+
+		// Check that the events match between monitors, even if they're in different queues.
+		let mut self_inner = self.inner.lock().unwrap();
+		let expected_pending: Vec<_> = provided.iter().chain(pending.iter()).cloned().collect();
+		assert_eq!(
+			self_inner.pending_monitor_events, expected_pending,
+			"Monitor events failed to round-trip serialization"
+		);
+
+		self_inner.provided_monitor_events = provided;
+		self_inner.pending_monitor_events = pending;
+	}
+
+	/// Used by test infra to check for unexpected pending monitor events.
+	#[cfg(any(test, feature = "_test_utils"))]
+	pub fn list_unacked_monitor_events(&self) -> Vec<(u128, MonitorEvent)> {
+		let inner = self.inner.lock().unwrap();
+		let mut events = Vec::new();
+		events.append(&mut inner.pending_monitor_events.clone());
+		events.append(&mut inner.provided_monitor_events.clone());
+		events
+	}
+
 	/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity.
 	///
 	/// For channels featuring anchor outputs, this method will also process [`BumpTransaction`]
@@ -2359,8 +2444,9 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		B: BroadcasterInterface,
 		F: FeeEstimator,
 		L: Logger,
+		ES: EntropySource,
 	>(
-		&self, broadcaster: &B, fee_estimator: &F, logger: &L,
+		&self, broadcaster: &B, fee_estimator: &F, logger: &L, entropy_source: &ES,
 	) {
 		let mut inner = self.inner.lock().unwrap();
 		let fee_estimator = LowerBoundedFeeEstimator::new(fee_estimator);
@@ -2371,6 +2457,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 			&fee_estimator,
 			&logger,
 			false,
+			entropy_source,
 		);
 	}
 
@@ -2398,29 +2485,36 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	///
 	/// [`get_outputs_to_watch`]: #method.get_outputs_to_watch
 	#[rustfmt::skip]
-	pub fn block_connected<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	pub fn block_connected<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&self,
 		header: &Header,
 		txdata: &TransactionData,
 		height: u32,
 		broadcaster: B,
 		fee_estimator: F,
 		logger: &L,
+		entropy_source: &ES,
 	) -> Vec<TransactionOutputs> {
 		let mut inner = self.inner.lock().unwrap();
 		let logger = WithChannelMonitor::from_impl(logger, &*inner, None);
 		inner.block_connected(
-			header, txdata, height, broadcaster, fee_estimator, &logger)
+			header, txdata, height, broadcaster, fee_estimator, &logger, entropy_source)
 	}
 
 	/// Determines if the disconnected block contained any transactions of interest and updates
 	/// appropriately.
-	pub fn blocks_disconnected<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	pub fn blocks_disconnected<
+		B: BroadcasterInterface,
+		F: FeeEstimator,
+		L: Logger,
+		ES: EntropySource,
+	>(
 		&self, fork_point: BlockLocator, broadcaster: B, fee_estimator: F, logger: &L,
+		entropy_source: &ES,
 	) {
 		let mut inner = self.inner.lock().unwrap();
 		let logger = WithChannelMonitor::from_impl(logger, &*inner, None);
-		inner.blocks_disconnected(fork_point, broadcaster, fee_estimator, &logger)
+		inner.blocks_disconnected(fork_point, broadcaster, fee_estimator, &logger, entropy_source)
 	}
 
 	/// Processes transactions confirmed in a block with the given header and height, returning new
@@ -2431,20 +2525,21 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	///
 	/// [`block_connected`]: Self::block_connected
 	#[rustfmt::skip]
-	pub fn transactions_confirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	pub fn transactions_confirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&self,
 		header: &Header,
 		txdata: &TransactionData,
 		height: u32,
 		broadcaster: B,
 		fee_estimator: F,
 		logger: &L,
+		entropy_source: &ES,
 	) -> Vec<TransactionOutputs> {
 		let bounded_fee_estimator = LowerBoundedFeeEstimator::new(fee_estimator);
 		let mut inner = self.inner.lock().unwrap();
 		let logger = WithChannelMonitor::from_impl(logger, &*inner, None);
 		inner.transactions_confirmed(
-			header, txdata, height, broadcaster, &bounded_fee_estimator, &logger)
+			header, txdata, height, broadcaster, &bounded_fee_estimator, &logger, entropy_source)
 	}
 
 	/// Processes a transaction that was reorganized out of the chain.
@@ -2454,18 +2549,19 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	///
 	/// [`blocks_disconnected`]: Self::blocks_disconnected
 	#[rustfmt::skip]
-	pub fn transaction_unconfirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	pub fn transaction_unconfirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&self,
 		txid: &Txid,
 		broadcaster: B,
 		fee_estimator: F,
 		logger: &L,
+		entropy_source: &ES,
 	) {
 		let bounded_fee_estimator = LowerBoundedFeeEstimator::new(fee_estimator);
 		let mut inner = self.inner.lock().unwrap();
 		let logger = WithChannelMonitor::from_impl(logger, &*inner, None);
 		inner.transaction_unconfirmed(
-			txid, broadcaster, &bounded_fee_estimator, &logger
+			txid, broadcaster, &bounded_fee_estimator, &logger, entropy_source
 		);
 	}
 
@@ -2477,19 +2573,20 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	///
 	/// [`block_connected`]: Self::block_connected
 	#[rustfmt::skip]
-	pub fn best_block_updated<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	pub fn best_block_updated<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&self,
 		header: &Header,
 		height: u32,
 		broadcaster: B,
 		fee_estimator: F,
 		logger: &L,
+		entropy_source: &ES,
 	) -> Vec<TransactionOutputs> {
 		let bounded_fee_estimator = LowerBoundedFeeEstimator::new(fee_estimator);
 		let mut inner = self.inner.lock().unwrap();
 		let logger = WithChannelMonitor::from_impl(logger, &*inner, None);
 		inner.best_block_updated(
-			header, height, broadcaster, &bounded_fee_estimator, &logger
+			header, height, broadcaster, &bounded_fee_estimator, &logger, entropy_source
 		)
 	}
 
@@ -2613,8 +2710,8 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 
 	/// Checks if the monitor is fully resolved. Resolved monitor is one that has claimed all of
 	/// its outputs and balances (i.e. [`Self::get_claimable_balances`] returns an empty set) and
-	/// which does not have any payment preimages for HTLCs which are still pending on other
-	/// channels.
+	/// which does not have any unacked [`MonitorEvent`]s which may contain resolutions for HTLCs
+	/// which are still pending on other channels.
 	///
 	/// Additionally may update state to track when the balances set became empty.
 	///
@@ -2631,38 +2728,44 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		let current_height = self.current_best_block().height;
 		let mut inner = self.inner.lock().unwrap();
 
+		// Unacked `MonitorEvent`s may contain the outbound-edge resolution of an inbound-edge HTLC
+		// that is present on another channel, in the case of a forward. Archiving while unacked
+		// monitor events are present could lead to the inbound edge channel not receiving the preimage
+		// in time (for forward claims), or to the inbound edge closing due to an HTLC timeout (for
+		// forward fails).
+		let no_unacked_monitor_events =
+			inner.pending_monitor_events.is_empty() && inner.provided_monitor_events.is_empty();
+
 		if inner.is_closed_without_updates()
 			&& is_all_funds_claimed
 			&& !inner.funding_spend_seen
 		{
 			// We closed the channel without ever advancing it and didn't have any funds in it. There's
-			// nothing for us to ever do with this monitor, so we archive it as soon as any pending
-			// `MonitorEvent`s have been processed. This may be necessary in the case that the monitor
+			// nothing for us to ever do with this monitor, so we archive it as soon as all pending
+			// `MonitorEvent`s have been acked. This may be necessary in the case that the monitor
 			// initiated the channel close -- archiving now may leave the `ChannelManager` with a
 			// `Channel` that has no corresponding monitor, which is not allowed on restart.
-			return (inner.pending_monitor_events.is_empty(), false);
+			return (no_unacked_monitor_events, false);
 		}
 
 		if is_all_funds_claimed && !inner.funding_spend_seen {
 			debug_assert!(false, "We should see funding spend by the time a monitor clears out");
 			is_all_funds_claimed = false;
 		}
 
-		// As long as HTLCs remain unresolved, they'll be present as a `Balance`. After that point,
-		// if they contained a preimage, an event will appear in `pending_monitor_events` which,
-		// once processed, implies the preimage exists in the corresponding inbound channel.
-		let preimages_not_needed_elsewhere = inner.pending_monitor_events.is_empty();
+		if !no_unacked_monitor_events {
+			return (false, false);
+		}
 
-		match (inner.balances_empty_height, is_all_funds_claimed, preimages_not_needed_elsewhere) {
-			(Some(balances_empty_height), true, true) => {
+		match (inner.balances_empty_height, is_all_funds_claimed) {
+			(Some(balances_empty_height), true) => {
 				// Claimed all funds, check if reached the blocks threshold.
 				(current_height >= balances_empty_height + ARCHIVAL_DELAY_BLOCKS, false)
 			},
-			(Some(_), false, _)|(Some(_), _, false) => {
-				// previously assumed we claimed all funds, but we have new funds to claim or
-				// preimages are suddenly needed (because of a duplicate-hash HTLC).
-				// This should never happen as once the `Balance`s and preimages are clear, we
-				// should never create new ones.
+			(Some(_), false) => {
+				// previously assumed we claimed all funds, but we have new funds to claim.
+				// This should never happen as once the `Balance`s are clear, we should never
+				// create new ones.
 				debug_assert!(false,
 					"Thought we were done claiming funds, but claimable_balances now has entries");
 				log_error!(logger,
@@ -2671,17 +2774,17 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 				inner.balances_empty_height = None;
 				(false, true)
 			},
-			(None, true, true) => {
-				// Claimed all funds and preimages can be deleted, but `balances_empty_height` is
-				// None. It is set to the current block height.
+			(None, true) => {
+				// Claimed all funds, but `balances_empty_height` is None. It is set to the current
+				// block height.
 				log_debug!(logger,
 					"ChannelMonitor funded at {} is now fully resolved. It will become archivable in {} blocks",
 					inner.get_funding_txo(), ARCHIVAL_DELAY_BLOCKS);
 				inner.balances_empty_height = Some(current_height);
 				(false, true)
 			},
-			(None, false, _)|(None, _, false) => {
-				// Have funds to claim or our preimages are still needed.
+			(None, false) => {
+				// Have funds to claim.
 				(false, false)
 			},
 		}
@@ -3916,9 +4019,9 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 
 	#[rustfmt::skip]
-	fn generate_claimable_outpoints_and_watch_outputs(
+	fn generate_claimable_outpoints_and_watch_outputs<ES: EntropySource>(
 		&mut self, generate_monitor_event_with_reason: Option<ClosureReason>,
-		require_funding_seen: bool,
+		require_funding_seen: bool, entropy_source: ES,
 	) -> (Vec<PackageTemplate>, Vec<TransactionOutputs>) {
 		let funding = get_confirmed_funding_scope!(self);
 		let holder_commitment_tx = &funding.current_holder_commitment_tx;
@@ -3939,7 +4042,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 				outpoint: funding_outpoint,
 				channel_id: self.channel_id,
 			};
-			self.pending_monitor_events.push(event);
+			push_monitor_event(&mut self.pending_monitor_events, event, entropy_source);
 		}
 
 		// Although we aren't signing the transaction directly here, the transaction will be signed
@@ -3990,16 +4093,16 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	/// See also [`ChannelMonitor::broadcast_latest_holder_commitment_txn`].
 	///
 	/// [`ChannelMonitor::broadcast_latest_holder_commitment_txn`]: crate::chain::channelmonitor::ChannelMonitor::broadcast_latest_holder_commitment_txn
-	pub(crate) fn queue_latest_holder_commitment_txn_for_broadcast<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	pub(crate) fn queue_latest_holder_commitment_txn_for_broadcast<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&mut self, broadcaster: &B, fee_estimator: &LowerBoundedFeeEstimator<F>, logger: &WithContext<L>,
-		require_funding_seen: bool,
+		require_funding_seen: bool, entropy_source: &ES,
 	) {
 		let reason = ClosureReason::HolderForceClosed {
 			broadcasted_latest_txn: Some(true),
 			message: "ChannelMonitor-initiated commitment transaction broadcast".to_owned(),
 		};
 		let (claimable_outpoints, _) =
-			self.generate_claimable_outpoints_and_watch_outputs(Some(reason), require_funding_seen);
+			self.generate_claimable_outpoints_and_watch_outputs(Some(reason), require_funding_seen, entropy_source);
 		// In manual-broadcast mode, if `require_funding_seen` is true and we have not yet observed
 		// the funding transaction on-chain, do not queue any transactions.
 		if require_funding_seen && self.is_manual_broadcast && !self.funding_seen_onchain {
@@ -4226,8 +4329,9 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 
 	#[rustfmt::skip]
-	fn update_monitor<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
-		&mut self, updates: &ChannelMonitorUpdate, broadcaster: &B, fee_estimator: &F, logger: &WithContext<L>
+	fn update_monitor<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
+		&mut self, updates: &ChannelMonitorUpdate, broadcaster: &B, fee_estimator: &F, logger: &WithContext<L>,
+		entropy_source: &ES,
 	) -> Result<(), ()> {
 		if self.latest_update_id == LEGACY_CLOSED_CHANNEL_UPDATE_ID && updates.update_id == LEGACY_CLOSED_CHANNEL_UPDATE_ID {
 			log_info!(logger, "Applying pre-0.1 post-force-closed update to monitor {} with {} change(s).",
@@ -4363,7 +4467,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 							log_trace!(logger, "Avoiding commitment broadcast, already detected confirmed spend onchain");
 							continue;
 						}
-						self.queue_latest_holder_commitment_txn_for_broadcast(broadcaster, &bounded_fee_estimator, logger, true);
+						self.queue_latest_holder_commitment_txn_for_broadcast(broadcaster, &bounded_fee_estimator, logger, true, entropy_source);
 					} else if !self.holder_tx_signed {
 						log_error!(logger, "WARNING: You have a potentially-unsafe holder commitment transaction available to broadcast");
 						log_error!(logger, "    in channel monitor!");
@@ -4409,7 +4513,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 						htlc_outputs.iter().filter_map(|(htlc, source)| {
 							source.as_ref().map(|s| (&**s, htlc.payment_hash, htlc.amount_msat))
 						}),
-						logger,
+						logger, entropy_source
 					);
 				},
 				ChannelMonitorUpdateStep::LatestCounterpartyCommitment {
@@ -4436,7 +4540,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					});
 					self.fail_htlcs_from_update_after_funding_spend(
 						nondust.chain(dust),
-						logger,
+						logger, entropy_source,
 					);
 				},
 				_ => {},
@@ -4500,9 +4604,9 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	/// Only truly new HTLCs (not present in any previously-known commitment) need to be failed
 	/// here. HTLCs that were already tracked by the monitor will be handled by the existing
 	/// `fail_unbroadcast_htlcs` logic when the spending transaction confirms.
-	fn fail_htlcs_from_update_after_funding_spend<'a, L: Logger>(
+	fn fail_htlcs_from_update_after_funding_spend<'a, L: Logger, ES: EntropySource>(
 		&mut self, htlcs: impl Iterator<Item = (&'a HTLCSource, PaymentHash, u64)>,
-		logger: &WithContext<L>,
+		logger: &WithContext<L>, entropy_source: &ES,
 	) {
 		let pending_spend_entry = self
 			.onchain_events_awaiting_threshold_conf
@@ -4566,12 +4670,16 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					"Failing HTLC from late counterparty commitment update immediately \
 					(funding spend already confirmed)"
 				);
-				self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
-					payment_hash,
-					payment_preimage: None,
-					source: source.clone(),
-					htlc_value_satoshis,
-				}));
+				push_monitor_event(
+					&mut self.pending_monitor_events,
+					MonitorEvent::HTLCEvent(HTLCUpdate {
+						payment_hash,
+						payment_preimage: None,
+						source: source.clone(),
+						htlc_value_satoshis,
+					}),
+					entropy_source,
+				);
 				self.htlcs_resolved_on_chain.push(IrrevocablyResolvedHTLC {
 					commitment_tx_output_idx: None,
 					resolving_txid: Some(confirmed_txid),
@@ -4636,9 +4744,20 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		&self.outputs_to_watch
 	}
 
-	fn get_and_clear_pending_monitor_events(&mut self) -> Vec<MonitorEvent> {
+	fn push_monitor_event<ES: EntropySource>(&mut self, event: MonitorEvent, entropy_source: ES) {
+		push_monitor_event(&mut self.pending_monitor_events, event, entropy_source);
+	}
+
+	fn ack_monitor_event(&mut self, event_id: u128) {
+		self.provided_monitor_events.retain(|(id, _)| *id != event_id);
+		// If this event was generated prior to a restart, it may be in this queue instead
+		self.pending_monitor_events.retain(|(id, _)| *id != event_id);
+	}
+
+	fn get_and_clear_pending_monitor_events(&mut self) -> Vec<(u128, MonitorEvent)> {
 		let mut ret = Vec::new();
 		mem::swap(&mut ret, &mut self.pending_monitor_events);
+		self.provided_monitor_events.extend(ret.iter().cloned());
 		ret
 	}
 
@@ -5484,29 +5603,30 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 
 	#[rustfmt::skip]
-	fn block_connected<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	fn block_connected<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&mut self, header: &Header, txdata: &TransactionData, height: u32, broadcaster: B,
-		fee_estimator: F, logger: &WithContext<L>,
+		fee_estimator: F, logger: &WithContext<L>, entropy_source: &ES
 	) -> Vec<TransactionOutputs> {
 		let bounded_fee_estimator = LowerBoundedFeeEstimator::new(fee_estimator);
-		self.transactions_confirmed(header, txdata, height, broadcaster, &bounded_fee_estimator, logger)
+		self.transactions_confirmed(header, txdata, height, broadcaster, &bounded_fee_estimator, logger, entropy_source)
 	}
 
 	#[rustfmt::skip]
-	fn best_block_updated<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	fn best_block_updated<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&mut self,
 		header: &Header,
 		height: u32,
 		broadcaster: B,
 		fee_estimator: &LowerBoundedFeeEstimator<F>,
 		logger: &WithContext<L>,
+		entropy_source: &ES,
 	) -> Vec<TransactionOutputs> {
 		let block_hash = header.block_hash();
 
 		if height > self.best_block.height {
 			self.best_block.update_for_new_tip(block_hash, height);
 			log_trace!(logger, "Connecting new block {} at height {}", block_hash, height);
-			self.block_confirmed(height, block_hash, vec![], vec![], vec![], &broadcaster, &fee_estimator, logger)
+			self.block_confirmed(height, block_hash, vec![], vec![], vec![], &broadcaster, &fee_estimator, logger, entropy_source)
 		} else if block_hash != self.best_block.block_hash {
 			self.best_block = BlockLocator::new(block_hash, height);
 			log_trace!(logger, "Best block re-orged, replaced with new block {} at height {}", block_hash, height);
@@ -5521,14 +5641,15 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 
 	#[rustfmt::skip]
-	fn transactions_confirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	fn transactions_confirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&mut self,
 		header: &Header,
 		txdata: &TransactionData,
 		height: u32,
 		broadcaster: B,
 		fee_estimator: &LowerBoundedFeeEstimator<F>,
 		logger: &WithContext<L>,
+		entropy_source: &ES,
 	) -> Vec<TransactionOutputs> {
 		let funding_seen_before = self.funding_seen_onchain;
 		let txn_matched = self.filter_block(txdata);
@@ -5703,7 +5824,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					);
 					log_info!(logger, "Channel closed by funding output spend in txid {txid}");
 					if !self.funding_spend_seen {
-						self.pending_monitor_events.push(MonitorEvent::CommitmentTxConfirmed(()));
+						self.push_monitor_event(MonitorEvent::CommitmentTxConfirmed(()), entropy_source);
 					}
 					self.funding_spend_seen = true;
 
@@ -5778,7 +5899,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					break;
 				}
 			}
-			self.is_resolving_htlc_output(&tx, height, &block_hash, logger);
+			self.is_resolving_htlc_output(&tx, height, &block_hash, logger, entropy_source);
 
 			// Note that if the funding transaction (or some arbitrary dependent of the funding
 			// transaction or some HTLC transaction) spends to the `destination_script` or
@@ -5793,12 +5914,12 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 
 		if should_broadcast_commitment {
 			let (mut claimables, mut outputs) =
-				self.generate_claimable_outpoints_and_watch_outputs(None, false);
+				self.generate_claimable_outpoints_and_watch_outputs(None, false, entropy_source);
 			claimable_outpoints.append(&mut claimables);
 			watch_outputs.append(&mut outputs);
 		}
 
-		self.block_confirmed(height, block_hash, txn_matched, watch_outputs, claimable_outpoints, &broadcaster, &fee_estimator, logger)
+		self.block_confirmed(height, block_hash, txn_matched, watch_outputs, claimable_outpoints, &broadcaster, &fee_estimator, logger, entropy_source)
 	}
 
 	/// Update state for new block(s)/transaction(s) confirmed. Note that the caller must update
@@ -5810,7 +5931,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	/// `conf_height` should be set to the height at which any new transaction(s)/block(s) were
 	/// confirmed at, even if it is not the current best height.
 	#[rustfmt::skip]
-	fn block_confirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	fn block_confirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&mut self,
 		conf_height: u32,
 		conf_hash: BlockHash,
@@ -5820,6 +5941,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		broadcaster: &B,
 		fee_estimator: &LowerBoundedFeeEstimator<F>,
 		logger: &WithContext<L>,
+		entropy_source: &ES,
 	) -> Vec<TransactionOutputs> {
 		log_trace!(logger, "Processing {} matched transactions for block at height {}.", txn_matched.len(), conf_height);
 		debug_assert!(self.best_block.height >= conf_height);
@@ -5830,7 +5952,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 			if let Some(payment_hash) = should_broadcast {
 				let reason = ClosureReason::HTLCsTimedOut { payment_hash: Some(payment_hash) };
 				let (mut new_outpoints, mut new_outputs) =
-					self.generate_claimable_outpoints_and_watch_outputs(Some(reason), false);
+					self.generate_claimable_outpoints_and_watch_outputs(Some(reason), false, entropy_source);
 				if !self.is_manual_broadcast || self.funding_seen_onchain {
 					claimable_outpoints.append(&mut new_outpoints);
 					watch_outputs.append(&mut new_outputs);
@@ -5880,12 +6002,12 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 
 					log_debug!(logger, "HTLC {} failure update in {} has got enough confirmations to be passed upstream",
 						&payment_hash, entry.txid);
-					self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
+					self.push_monitor_event(MonitorEvent::HTLCEvent(HTLCUpdate {
 						payment_hash,
 						payment_preimage: None,
 						source,
 						htlc_value_satoshis,
-					}));
+					}), entropy_source);
 					self.htlcs_resolved_on_chain.push(IrrevocablyResolvedHTLC {
 						commitment_tx_output_idx,
 						resolving_txid: Some(entry.txid),
@@ -5975,8 +6097,8 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 				if inbound_htlc_expiry > max_expiry_height {
 					continue;
 				}
-				let duplicate_event = self.pending_monitor_events.iter().any(
-					|update| if let &MonitorEvent::HTLCEvent(ref upd) = update {
+				let duplicate_event = self.pending_monitor_events.iter().chain(self.provided_monitor_events.iter())
+					.any(|(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update {
 						upd.source == *source
 					} else { false });
 				if duplicate_event {
@@ -5989,12 +6111,12 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					log_error!(logger, "Failing back HTLC {} upstream to preserve the \
 						channel as the forward HTLC hasn't resolved and our backward HTLC \
 						expires soon at {}", log_bytes!(htlc.payment_hash.0), inbound_htlc_expiry);
-					self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
+					push_monitor_event(&mut self.pending_monitor_events, MonitorEvent::HTLCEvent(HTLCUpdate {
 						source: source.clone(),
 						payment_preimage: None,
 						payment_hash: htlc.payment_hash,
 						htlc_value_satoshis: htlc.amount_msat / 1000,
-					}));
+					}), entropy_source);
 				}
 			}
 		}
@@ -6033,8 +6155,9 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 
 	#[rustfmt::skip]
-	fn blocks_disconnected<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
-		&mut self, fork_point: BlockLocator, broadcaster: B, fee_estimator: F, logger: &WithContext<L>
+	fn blocks_disconnected<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
+		&mut self, fork_point: BlockLocator, broadcaster: B, fee_estimator: F, logger: &WithContext<L>,
+		entropy_source: &ES,
 	) {
 		let new_height = fork_point.height;
 		log_trace!(logger, "Block(s) disconnected to height {}", new_height);
@@ -6079,19 +6202,20 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		// Only attempt to broadcast the new commitment after the `block_disconnected` call above so that
 		// it doesn't get removed from the set of pending claims.
 		if should_broadcast_commitment {
-			self.queue_latest_holder_commitment_txn_for_broadcast(&broadcaster, &bounded_fee_estimator, logger, true);
+			self.queue_latest_holder_commitment_txn_for_broadcast(&broadcaster, &bounded_fee_estimator, logger, true, entropy_source);
 		}
 
 		self.best_block = fork_point;
 	}
 
 	#[rustfmt::skip]
-	fn transaction_unconfirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger>(
+	fn transaction_unconfirmed<B: BroadcasterInterface, F: FeeEstimator, L: Logger, ES: EntropySource>(
 		&mut self,
 		txid: &Txid,
 		broadcaster: B,
 		fee_estimator: &LowerBoundedFeeEstimator<F>,
 		logger: &WithContext<L>,
+		entropy_source: &ES,
 	) {
 		let mut removed_height = None;
 		for entry in self.onchain_events_awaiting_threshold_conf.iter() {
@@ -6141,7 +6265,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		// Only attempt to broadcast the new commitment after the `transaction_unconfirmed` call above so
 		//  that it doesn't get removed from the set of pending claims.
 		if should_broadcast_commitment {
-			self.queue_latest_holder_commitment_txn_for_broadcast(&broadcaster, fee_estimator, logger, true);
+			self.queue_latest_holder_commitment_txn_for_broadcast(&broadcaster, fee_estimator, logger, true, entropy_source);
 		}
 	}
 
@@ -6286,8 +6410,9 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	/// Check if any transaction broadcasted is resolving HTLC output by a success or timeout on a holder
 	/// or counterparty commitment tx, if so send back the source, preimage if found and payment_hash of resolved HTLC
 	#[rustfmt::skip]
-	fn is_resolving_htlc_output<L: Logger>(
+	fn is_resolving_htlc_output<L: Logger, ES: EntropySource>(
 		&mut self, tx: &Transaction, height: u32, block_hash: &BlockHash, logger: &WithContext<L>,
+		entropy_source: &ES,
 	) {
 		let funding_spent = get_confirmed_funding_scope!(self);
 
@@ -6464,20 +6589,22 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					self.counterparty_fulfilled_htlcs.insert(SentHTLCId::from_source(&source), payment_preimage);
 					// We may have already queued a failure of this HTLC upstream because the upstream HTLC
 					// was about to expire while this one was still unresolved on chain. The counterparty
-					// has now revealed the preimage instead, and events still queued here have not been
-					// provided to anyone yet, so drop the failure in favor of claiming upstream.
-					self.pending_monitor_events.retain(|update| match update {
+					// has now revealed the preimage instead, so drop the failure in favor of claiming
+					// upstream.
+					let not_htlc_fail = |(_, update): &(u128, MonitorEvent)| match update {
 						MonitorEvent::HTLCEvent(upd) => upd.source != source || upd.payment_preimage.is_some(),
 						_ => true,
-					});
-					if !self.pending_monitor_events.iter().any(
-						|update| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) {
-						self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
+					};
+					self.pending_monitor_events.retain(not_htlc_fail);
+					self.provided_monitor_events.retain(not_htlc_fail);
+					if !self.pending_monitor_events.iter().chain(self.provided_monitor_events.iter()).any(
+						|(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) {
+						push_monitor_event(&mut self.pending_monitor_events, MonitorEvent::HTLCEvent(HTLCUpdate {
 							source,
 							payment_preimage: Some(payment_preimage),
 							payment_hash,
 							htlc_value_satoshis: amount_msat / 1000,
-						}));
+						}), entropy_source);
 					}
 				} else {
 					self.onchain_events_awaiting_threshold_conf.retain(|ref entry| {
@@ -6578,33 +6705,44 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 }
 
-impl<Signer: EcdsaChannelSigner, T: BroadcasterInterface, F: FeeEstimator, L: Logger> chain::Listen
-	for (ChannelMonitor<Signer>, T, F, L)
+impl<
+		Signer: EcdsaChannelSigner,
+		T: BroadcasterInterface,
+		F: FeeEstimator,
+		L: Logger,
+		ES: EntropySource,
+	> chain::Listen for (ChannelMonitor<Signer>, T, F, L, ES)
 {
 	fn filtered_block_connected(&self, header: &Header, txdata: &TransactionData, height: u32) {
-		self.0.block_connected(header, txdata, height, &self.1, &self.2, &self.3);
+		self.0.block_connected(header, txdata, height, &self.1, &self.2, &self.3, &self.4);
 	}
 
 	fn blocks_disconnected(&self, fork_point: BlockLocator) {
-		self.0.blocks_disconnected(fork_point, &self.1, &self.2, &self.3);
+		self.0.blocks_disconnected(fork_point, &self.1, &self.2, &self.3, &self.4);
 	}
 }
 
-impl<Signer: EcdsaChannelSigner, M, T: BroadcasterInterface, F: FeeEstimator, L: Logger>
-	chain::Confirm for (M, T, F, L)
+impl<
+		Signer: EcdsaChannelSigner,
+		M,
+		T: BroadcasterInterface,
+		F: FeeEstimator,
+		L: Logger,
+		ES: EntropySource,
+	> chain::Confirm for (M, T, F, L, ES)
 where
 	M: Deref<Target = ChannelMonitor<Signer>>,
 {
 	fn transactions_confirmed(&self, header: &Header, txdata: &TransactionData, height: u32) {
-		self.0.transactions_confirmed(header, txdata, height, &self.1, &self.2, &self.3);
+		self.0.transactions_confirmed(header, txdata, height, &self.1, &self.2, &self.3, &self.4);
 	}
 
 	fn transaction_unconfirmed(&self, txid: &Txid) {
-		self.0.transaction_unconfirmed(txid, &self.1, &self.2, &self.3);
+		self.0.transaction_unconfirmed(txid, &self.1, &self.2, &self.3, &self.4);
 	}
 
 	fn best_block_updated(&self, header: &Header, height: u32) {
-		self.0.best_block_updated(header, height, &self.1, &self.2, &self.3);
+		self.0.best_block_updated(header, height, &self.1, &self.2, &self.3, &self.4);
 	}
 
 	fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option<BlockHash>)> {
@@ -6770,16 +6908,16 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			}
 		}
 
-		let pending_monitor_events_len: u64 = Readable::read(reader)?;
-		let mut pending_monitor_events = Some(
-			Vec::with_capacity(cmp::min(pending_monitor_events_len as usize, MAX_ALLOC_SIZE / (32 + 8*3))));
-		for _ in 0..pending_monitor_events_len {
+		let pending_monitor_events_legacy_len: u64 = Readable::read(reader)?;
+		let mut pending_monitor_events_legacy = Some(
+			Vec::with_capacity(cmp::min(pending_monitor_events_legacy_len as usize, MAX_ALLOC_SIZE / (32 + 8*3))));
+		for _ in 0..pending_monitor_events_legacy_len {
 			let ev = match <u8 as Readable>::read(reader)? {
 				0 => MonitorEvent::HTLCEvent(Readable::read(reader)?),
 				1 => MonitorEvent::HolderForceClosed(outpoint),
 				_ => return Err(DecodeError::InvalidValue)
 			};
-			pending_monitor_events.as_mut().unwrap().push(ev);
+			pending_monitor_events_legacy.as_mut().unwrap().push(ev);
 		}
 
 		let pending_events_len: u64 = Readable::read(reader)?;
@@ -6844,10 +6982,11 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 		let mut current_funding_contribution = None;
 		let mut funding_tx_confirmed_in = None;
 		let mut alternative_funding_confirmed_block = None;
+		let mut pending_mon_evs_with_ids: Option<Vec<ReadableIdMonitorEvent>> = None;
 		read_tlv_fields!(reader, {
 			(1, funding_spend_confirmed, option),
 			(3, htlcs_resolved_on_chain, optional_vec),
-			(5, pending_monitor_events, optional_vec),
+			(5, pending_monitor_events_legacy, optional_vec),
 			(7, funding_spend_seen, option),
 			(9, counterparty_node_id, option),
 			(11, confirmed_commitment_tx_counterparty_output, option),
@@ -6870,6 +7009,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			(41, current_funding_contribution, option),
 			(43, funding_tx_confirmed_in, option),
 			(45, alternative_funding_confirmed_block, option),
+			(47, pending_mon_evs_with_ids, optional_vec),
 		});
 		if let Some(previous_blocks) = best_block_previous_blocks {
 			best_block.previous_blocks = previous_blocks;
@@ -6905,14 +7045,25 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 
 		// `HolderForceClosedWithInfo` replaced `HolderForceClosed` in v0.0.122. If we have both
 		// events, we can remove the `HolderForceClosed` event and just keep the `HolderForceClosedWithInfo`.
-		if let Some(ref mut pending_monitor_events) = pending_monitor_events {
-			if pending_monitor_events.iter().any(|e| matches!(e, MonitorEvent::HolderForceClosed(_))) &&
-				pending_monitor_events.iter().any(|e| matches!(e, MonitorEvent::HolderForceClosedWithInfo { .. }))
+		if let Some(ref mut evs) = pending_monitor_events_legacy {
+			if evs.iter().any(|e| matches!(e, MonitorEvent::HolderForceClosed(_))) &&
+				evs.iter().any(|e| matches!(e, MonitorEvent::HolderForceClosedWithInfo { .. }))
 			{
-				pending_monitor_events.retain(|e| !matches!(e, MonitorEvent::HolderForceClosed(_)));
+				evs.retain(|e| !matches!(e, MonitorEvent::HolderForceClosed(_)));
 			}
 		}
 
+		let pending_monitor_events: Vec<(u128, MonitorEvent)> =
+			if let Some(pending_mon_evs_with_ids) = pending_mon_evs_with_ids {
+				pending_mon_evs_with_ids.into_iter().map(|ev| (ev.0, ev.1)).collect()
+			} else if let Some(events) = pending_monitor_events_legacy {
+				events.into_iter()
+					.map(|ev| (random_monitor_event_id(entropy_source), ev))
+					.collect()
+			} else {
+				Vec::new()
+			};
+
 		let channel_parameters = channel_parameters.unwrap_or_else(|| {
 			onchain_tx_handler.channel_parameters().clone()
 		});
@@ -7031,7 +7182,8 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			current_holder_commitment_number,
 
 			payment_preimages,
-			pending_monitor_events: pending_monitor_events.unwrap(),
+			pending_monitor_events,
+			provided_monitor_events: Vec::new(),
 			pending_events,
 			is_processing_pending_events: false,
 
@@ -7081,6 +7233,22 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 	}
 }
 
+/// Deserialization wrapper for reading a `(u128, MonitorEvent)`.
+/// Necessary because we can't deserialize a `(Readable, MaybeReadable)` tuple due to trait
+/// conflicts.
+struct ReadableIdMonitorEvent(u128, MonitorEvent);
+
+impl MaybeReadable for ReadableIdMonitorEvent {
+	fn read<R: io::Read>(reader: &mut R) -> Result<Option<Self>, DecodeError> {
+		let id: u128 = Readable::read(reader)?;
+		let event_opt: Option<MonitorEvent> = MaybeReadable::read(reader)?;
+		match event_opt {
+			Some(ev) => Ok(Some(ReadableIdMonitorEvent(id, ev))),
+			None => Ok(None),
+		}
+	}
+}
+
 #[cfg(test)]
 pub(super) fn dummy_monitor<S: EcdsaChannelSigner + 'static>(
 	channel_id: ChannelId,
@@ -7278,7 +7446,7 @@ mod tests {
 
 		let broadcaster = TestBroadcaster::with_blocks(Arc::clone(&nodes[1].blocks));
 		assert!(
-			pre_update_monitor.update_monitor(&replay_update, &&broadcaster, &&chanmon_cfgs[1].fee_estimator, &nodes[1].logger)
+			pre_update_monitor.update_monitor(&replay_update, &&broadcaster, &&chanmon_cfgs[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager)
 			.is_err());
 
 		// Even though we error'd on the first update, we should still have generated an HTLC claim

### lightning/src/chain/mod.rs
@@ -18,6 +18,7 @@ use bitcoin::network::Network;
 use bitcoin::script::{Script, ScriptBuf};
 use bitcoin::secp256k1::PublicKey;
 
+use crate::chain::chainmonitor::MonitorEventSource;
 use crate::chain::channelmonitor::{
 	ChannelMonitor, ChannelMonitorUpdate, MonitorEvent, ANTI_REORG_DELAY,
 };
@@ -421,6 +422,10 @@ pub trait Watch<ChannelSigner: EcdsaChannelSigner> {
 	/// Returns any monitor events since the last call. Subsequent calls must only return new
 	/// events.
 	///
+	/// Each event comes with a corresponding id. Once the event is processed, call
+	/// [`Watch::ack_monitor_event`] with the corresponding id and channel id. Unacknowledged events
+	/// will be re-provided by this method after startup.
+	///
 	/// Note that after any block- or transaction-connection calls to a [`ChannelMonitor`], no
 	/// further events may be returned here until the [`ChannelMonitor`] has been fully persisted
 	/// to disk.
@@ -429,7 +434,16 @@ pub trait Watch<ChannelSigner: EcdsaChannelSigner> {
 	/// [`MonitorEvent::Completed`] here, see [`ChannelMonitorUpdateStatus::InProgress`].
 	fn release_pending_monitor_events(
 		&self,
-	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)>;
+	) -> Vec<(OutPoint, ChannelId, Vec<(u128, MonitorEvent)>, PublicKey)>;
+
+	/// Acknowledges and removes a [`MonitorEvent`] previously returned by
+	/// [`Watch::release_pending_monitor_events`], keyed by the event's [`MonitorEventSource`].
+	///
+	/// Once persistently acknowledged, the event will no longer be returned by future calls to
+	/// [`Watch::release_pending_monitor_events`] and will not be replayed on restart.
+	///
+	/// Events may be acknowledged in any order.
+	fn ack_monitor_event(&self, source: MonitorEventSource);
 }
 
 impl<ChannelSigner: EcdsaChannelSigner, T: Watch<ChannelSigner> + ?Sized, W: Deref<Target = T>>
@@ -449,9 +463,13 @@ impl<ChannelSigner: EcdsaChannelSigner, T: Watch<ChannelSigner> + ?Sized, W: Der
 
 	fn release_pending_monitor_events(
 		&self,
-	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
+	) -> Vec<(OutPoint, ChannelId, Vec<(u128, MonitorEvent)>, PublicKey)> {
 		self.deref().release_pending_monitor_events()
 	}
+
+	fn ack_monitor_event(&self, source: MonitorEventSource) {
+		self.deref().ack_monitor_event(source)
+	}
 }
 
 /// The `Filter` trait defines behavior for indicating chain activity of interest pertaining to

### lightning/src/ln/chanmon_update_fail_tests.rs
@@ -5021,7 +5021,7 @@ fn native_async_persist() {
 	let completed_persist = async_chain_monitor.release_pending_monitor_events();
 	assert_eq!(completed_persist.len(), 1);
 	assert_eq!(completed_persist[0].2.len(), 1);
-	assert!(matches!(completed_persist[0].2[0], MonitorEvent::Completed { .. }));
+	assert!(matches!(completed_persist[0].2[0].1, MonitorEvent::Completed { .. }));
 
 	// Now test two async `ChannelMonitorUpdate`s in flight at once, completing them in-order but
 	// separately.
@@ -5069,7 +5069,7 @@ fn native_async_persist() {
 	let completed_persist = async_chain_monitor.release_pending_monitor_events();
 	assert_eq!(completed_persist.len(), 1);
 	assert_eq!(completed_persist[0].2.len(), 1);
-	assert!(matches!(completed_persist[0].2[0], MonitorEvent::Completed { .. }));
+	assert!(matches!(completed_persist[0].2[0].1, MonitorEvent::Completed { .. }));
 
 	// Finally, test two async `ChanelMonitorUpdate`s in flight at once, completing them
 	// out-of-order and ensuring that no `MonitorEvent::Completed` is generated until they are both
@@ -5115,7 +5115,7 @@ fn native_async_persist() {
 	let completed_persist = async_chain_monitor.release_pending_monitor_events();
 	assert_eq!(completed_persist.len(), 1);
 	assert_eq!(completed_persist[0].2.len(), 1);
-	if let MonitorEvent::Completed { monitor_update_id, .. } = &completed_persist[0].2[0] {
+	if let (_, MonitorEvent::Completed { monitor_update_id, .. }) = &completed_persist[0].2[0] {
 		assert_eq!(*monitor_update_id, 4);
 	} else {
 		panic!();

### lightning/src/ln/channelmanager.rs
@@ -42,6 +42,7 @@ use crate::chain::chaininterface::{
 	BroadcasterInterface, ConfirmationTarget, FeeEstimator, LowerBoundedFeeEstimator,
 	TransactionType,
 };
+use crate::chain::chainmonitor::MonitorEventSource;
 use crate::chain::channelmonitor::{
 	ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, MonitorEvent,
 	WithChannelMonitor, ANTI_REORG_DELAY, CLTV_CLAIM_BUFFER, HTLC_FAIL_BACK_BUFFER,
@@ -14455,7 +14456,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		for (funding_outpoint, channel_id, mut monitor_events, counterparty_node_id) in
 			pending_monitor_events.drain(..)
 		{
-			for monitor_event in monitor_events.drain(..) {
+			for (event_id, monitor_event) in monitor_events.drain(..) {
+				let monitor_event_source = MonitorEventSource { event_id, channel_id };
 				match monitor_event {
 					MonitorEvent::HTLCEvent(htlc_update) => {
 						needs_persist = true;
@@ -14506,6 +14508,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								completion_update,
 							);
 						}
+						self.chain_monitor.ack_monitor_event(monitor_event_source);
 					},
 					MonitorEvent::HolderForceClosed(_)
 					| MonitorEvent::HolderForceClosedWithInfo { .. } => {
@@ -14540,6 +14543,9 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								failed_channels.push((Err(e), counterparty_node_id));
 							}
 						}
+						// Channel close monitor events do not need to be replayed on startup because we
+						// already check the monitors to see if the channel is closed.
+						self.chain_monitor.ack_monitor_event(monitor_event_source);
 					},
 					MonitorEvent::CommitmentTxConfirmed(_) => {
 						needs_persist = true;
@@ -14562,13 +14568,17 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								failed_channels.push((Err(e), counterparty_node_id));
 							}
 						}
+						// Channel close monitor events do not need to be replayed on startup because we
+						// already check the monitors to see if the channel is closed.
+						self.chain_monitor.ack_monitor_event(monitor_event_source);
 					},
 					MonitorEvent::Completed { channel_id, monitor_update_id, .. } => {
 						needs_persist |= self.channel_monitor_updated(
 							&channel_id,
 							Some(monitor_update_id),
 							&counterparty_node_id,
 						);
+						self.chain_monitor.ack_monitor_event(monitor_event_source);
 					},
 				}
 			}

### lightning/src/ln/functional_test_utils.rs
@@ -846,6 +846,18 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
 				panic!("Had excess RAA blockers on node {}: {:?}", self.logger.id, raa_blockers);
 			}
 
+			for channel_id in self.chain_monitor.chain_monitor.list_monitors() {
+				let monitor = self.chain_monitor.chain_monitor.get_monitor(channel_id).unwrap();
+				let unacked_monitor_events = monitor.list_unacked_monitor_events();
+				if !unacked_monitor_events.is_empty() {
+					panic!(
+						"{} (channel {channel_id:?}) had {} unacked monitor events at drop: {unacked_monitor_events:#?}",
+						self.logger.id,
+						unacked_monitor_events.len()
+					);
+				}
+			}
+
 			// Check that if we serialize the network graph, we can deserialize it again.
 			let network_graph = {
 				let mut w = test_utils::TestVecWriter(Vec::new());

### lightning/src/ln/functional_tests.rs
@@ -10078,6 +10078,7 @@ fn do_test_manual_broadcast_skips_commitment_until_funding(
 			&nodes[0].tx_broadcaster,
 			&nodes[0].fee_estimator,
 			&nodes[0].logger,
+			&nodes[0].keys_manager,
 		);
 	} else {
 		mine_transaction(&nodes[0], &funding_tx);

### lightning/src/ln/monitor_tests.rs
@@ -339,7 +339,8 @@ fn archive_monitor_with_pending_closure_event() {
 	// Broadcast nodes[1]'s commitment transaction via the `ChannelMonitor` directly, so that the
 	// only indication of the closure the `ChannelManager` will ever get is the `MonitorEvent`.
 	get_monitor!(nodes[1], chan_id).broadcast_latest_holder_commitment_txn(
-		&nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger
+		&nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger,
+		&nodes[1].keys_manager,
 	);
 	let commitment_tx = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
 	assert_eq!(commitment_tx.len(), 1);
@@ -3245,7 +3246,8 @@ fn do_test_monitor_claims_with_random_signatures(keyed_anchors: bool, p2a_anchor
 	};
 
 	get_monitor!(closing_node, chan_id).broadcast_latest_holder_commitment_txn(
-		&closing_node.tx_broadcaster, &closing_node.fee_estimator, &closing_node.logger
+		&closing_node.tx_broadcaster, &closing_node.fee_estimator, &closing_node.logger,
+		&closing_node.keys_manager,
 	);
 	if keyed_anchors || p2a_anchor {
 		handle_bump_close_event(&closing_node);
@@ -3403,7 +3405,7 @@ fn test_update_replay_panics() {
 	// Update `monitor` until there's just one normal updates, an FC update, and a post-FC claim
 	// update pending
 	for update in updates.drain(..updates.len() - 4) {
-		monitor.update_monitor(&update, &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger).unwrap();
+		monitor.update_monitor(&update, &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager).unwrap();
 	}
 	assert_eq!(updates.len(), 4);
 	assert!(matches!(updates[1].updates[0], ChannelMonitorUpdateStep::ChannelForceClosed { .. }));
@@ -3413,31 +3415,31 @@ fn test_update_replay_panics() {
 	// Ensure applying the force-close update skipping the last normal update fails
 	let poisoned_monitor = monitor.clone();
 	std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-		let _ = poisoned_monitor.update_monitor(&updates[1], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger);
+		let _ = poisoned_monitor.update_monitor(&updates[1], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager);
 		// We should panic, rather than returning an error here.
 	})).unwrap_err();
 
 	// Then apply the last normal and force-close update and make sure applying the preimage
 	// updates out-of-order fails.
-	monitor.update_monitor(&updates[0], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger).unwrap();
-	monitor.update_monitor(&updates[1], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger).unwrap();
+	monitor.update_monitor(&updates[0], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager).unwrap();
+	monitor.update_monitor(&updates[1], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager).unwrap();
 
 	let poisoned_monitor = monitor.clone();
 	std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-		let _ = poisoned_monitor.update_monitor(&updates[3], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger);
+		let _ = poisoned_monitor.update_monitor(&updates[3], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager);
 		// We should panic, rather than returning an error here.
 	})).unwrap_err();
 
 	// Make sure re-applying the force-close update fails
 	let poisoned_monitor = monitor.clone();
 	std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-		let _ = poisoned_monitor.update_monitor(&updates[1], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger);
+		let _ = poisoned_monitor.update_monitor(&updates[1], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager);
 		// We should panic, rather than returning an error here.
 	})).unwrap_err();
 
 	// ...and finally ensure that applying all the updates succeeds.
-	monitor.update_monitor(&updates[2], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger).unwrap();
-	monitor.update_monitor(&updates[3], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger).unwrap();
+	monitor.update_monitor(&updates[2], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager).unwrap();
+	monitor.update_monitor(&updates[3], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger, &nodes[1].keys_manager).unwrap();
 }
 
 #[test]
@@ -3675,267 +3677,6 @@ fn test_lost_preimage_monitor_events() {
 	do_test_lost_preimage_monitor_events(false, true);
 }
 
-#[derive(PartialEq)]
-enum CommitmentType {
-	RevokedCounterparty,
-	LatestCounterparty,
-	PreviousCounterparty,
-	LocalWithoutLastHTLC,
-	LocalWithLastHTLC,
-}
-
-fn do_test_lost_timeout_monitor_events(confirm_tx: CommitmentType, dust_htlcs: bool, p2a_anchor: bool) {
-	// `MonitorEvent`s aren't delivered to the `ChannelManager` in a durable fashion - if the
-	// `ChannelManager` fetches the pending `MonitorEvent`s, then the `ChannelMonitor` gets
-	// persisted (i.e. due to a block update) then the node crashes, prior to persisting the
-	// `ChannelManager` again, the `MonitorEvent` and its effects on the `ChannelManger` will be
-	// lost. This isn't likely in a sync persist environment, but in an async one this could be an
-	// issue.
-	//
-	// Note that this is only an issue for closed channels - `MonitorEvent`s only inform the
-	// `ChannelManager` that a channel is closed (which the `ChannelManager` will learn on startup
-	// or when it next tries to advance the channel state), that `ChannelMonitorUpdate` writes
-	// completed (which the `ChannelManager` will detect on startup), or that HTLCs resolved
-	// on-chain post closure. Of the three, only the last is problematic to lose prior to a reload.
-	//
-	// Here we test that losing `MonitorEvent`s that contain HTLC resolution via timeouts does not
-	// cause us to lose a `PaymentFailed` event.
-	let mut cfg = test_default_channel_config();
-	cfg.channel_handshake_config.negotiate_anchors_zero_fee_htlc_tx = true;
-	cfg.channel_handshake_config.negotiate_anchor_zero_fee_commitments = p2a_anchor;
-	let cfgs = [Some(cfg.clone()), Some(cfg.clone()), Some(cfg.clone())];
-
-	let chanmon_cfgs = create_chanmon_cfgs(3);
-	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
-	let persister;
-	let new_chain_mon;
-	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &cfgs);
-	let node_b_reload;
-	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
-
-	provide_anchor_reserves(&nodes);
-
-	let node_a_id = nodes[0].node.get_our_node_id();
-	let node_b_id = nodes[1].node.get_our_node_id();
-	let node_c_id = nodes[2].node.get_our_node_id();
-
-	let chan_a = create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 1_000_000, 0).2;
-	let chan_b = create_announced_chan_between_nodes_with_value(&nodes, 1, 2, 1_000_000, 0).2;
-
-	// Ensure all nodes are at the same height
-	let node_max_height =
-		nodes.iter().map(|node| node.blocks.lock().unwrap().len()).max().unwrap() as u32;
-	connect_blocks(&nodes[0], node_max_height - nodes[0].best_block_info().1);
-	connect_blocks(&nodes[1], node_max_height - nodes[1].best_block_info().1);
-	connect_blocks(&nodes[2], node_max_height - nodes[2].best_block_info().1);
-
-	send_payment(&nodes[0], &[&nodes[1], &nodes[2]], 25_000_000);
-
-	let cs_revoked_commit = get_local_commitment_txn!(nodes[2], chan_b);
-	assert_eq!(cs_revoked_commit.len(), 1);
-
-	let amt = if dust_htlcs { 1_000 } else { 10_000_000 };
-	let (_, hash_a, ..) = route_payment(&nodes[0], &[&nodes[1], &nodes[2]], amt);
-
-	let cs_previous_commit = get_local_commitment_txn!(nodes[2], chan_b);
-	assert_eq!(cs_previous_commit.len(), 1);
-
-	let (route, hash_b, _, payment_secret_b) =
-		get_route_and_payment_hash!(nodes[1], nodes[2], amt);
-	let onion = RecipientOnionFields::secret_only(payment_secret_b, amt);
-	nodes[1].node.send_payment_with_route(route, hash_b, onion, PaymentId(hash_b.0)).unwrap();
-	check_added_monitors(&nodes[1], 1);
-
-	let updates = get_htlc_update_msgs(&nodes[1], &node_c_id);
-	nodes[2].node.handle_update_add_htlc(node_b_id, &updates.update_add_htlcs[0]);
-	nodes[2].node.handle_commitment_signed_batch_test(node_b_id, &updates.commitment_signed);
-	check_added_monitors(&nodes[2], 1);
-
-	let (cs_raa, cs_cs) = get_revoke_commit_msgs(&nodes[2], &node_b_id);
-	if confirm_tx == CommitmentType::LocalWithLastHTLC {
-		// Only deliver the last RAA + CS if we need to update the local commitment with the third
-		// HTLC.
-		nodes[1].node.handle_revoke_and_ack(node_c_id, &cs_raa);
-		check_added_monitors(&nodes[1], 1);
-		nodes[1].node.handle_commitment_signed_batch_test(node_c_id, &cs_cs);
-		check_added_monitors(&nodes[1], 1);
-
-		let _bs_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, node_c_id);
-	}
-
-	nodes[1].node.peer_disconnected(nodes[2].node.get_our_node_id());
-	nodes[2].node.peer_disconnected(nodes[1].node.get_our_node_id());
-
-	// Force-close the channel, confirming a commitment transaction then letting C claim the HTLCs.
-	let message = "Closed".to_owned();
-	nodes[2]
-		.node
-		.force_close_broadcasting_latest_txn(&chan_b, &node_b_id, message.clone())
-		.unwrap();
-	check_added_monitors(&nodes[2], 1);
-	let c_reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
-	check_closed_event(&nodes[2], 1, c_reason, &[node_b_id], 1_000_000);
-	check_closed_broadcast(&nodes[2], 1, false);
-
-	handle_bump_events(&nodes[2], true, 0);
-	let cs_commit_tx = nodes[2].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
-	assert_eq!(cs_commit_tx.len(), if p2a_anchor { 2 } else { 1 });
-
-	let message = "Closed".to_owned();
-	nodes[1]
-		.node
-		.force_close_broadcasting_latest_txn(&chan_b, &node_c_id, message.clone())
-		.unwrap();
-	check_added_monitors(&nodes[1], 1);
-	let b_reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
-	check_closed_event(&nodes[1], 1, b_reason, &[node_c_id], 1_000_000);
-	check_closed_broadcast(&nodes[1], 1, false);
-
-	handle_bump_events(&nodes[1], true, 0);
-	let bs_commit_tx = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
-	assert_eq!(bs_commit_tx.len(), if p2a_anchor { 2 } else { 1 });
-
-	let selected_commit_tx = match confirm_tx {
-		CommitmentType::RevokedCounterparty => &cs_revoked_commit[0],
-		CommitmentType::PreviousCounterparty => &cs_previous_commit[0],
-		CommitmentType::LatestCounterparty => &cs_commit_tx[0],
-		CommitmentType::LocalWithoutLastHTLC|CommitmentType::LocalWithLastHTLC => &bs_commit_tx[0],
-	};
-
-	mine_transaction(&nodes[1], selected_commit_tx);
-	// If the block gets connected first we may re-broadcast B's commitment transaction before
-	// seeing the C's confirm. In any case, if we confirmed the revoked counterparty commitment
-	// transaction, we want to go ahead and confirm the spend of it.
-	let bs_transactions = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
-	if confirm_tx == CommitmentType::RevokedCounterparty {
-		assert!(bs_transactions.len() == 1 || bs_transactions.len() == 2);
-		mine_transaction(&nodes[1], bs_transactions.last().unwrap());
-	} else {
-		assert!(bs_transactions.len() == 1 || bs_transactions.len() == 0);
-	}
-
-	connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
-	let mut events = nodes[1].chain_monitor.chain_monitor.get_and_clear_pending_events();
-	match confirm_tx {
-		CommitmentType::LocalWithoutLastHTLC|CommitmentType::LocalWithLastHTLC => {
-			assert_eq!(events.len(), 0, "{events:?}");
-		},
-		CommitmentType::PreviousCounterparty|CommitmentType::LatestCounterparty => {
-			assert_eq!(events.len(), 1, "{events:?}");
-			match events[0] {
-				Event::SpendableOutputs { .. } => {},
-				_ => panic!("Unexpected event {events:?}"),
-			}
-		},
-		CommitmentType::RevokedCounterparty => {
-			assert_eq!(events.len(), 2, "{events:?}");
-			for event in events {
-				match event {
-					Event::SpendableOutputs { .. } => {},
-					_ => panic!("Unexpected event {event:?}"),
-				}
-			}
-		},
-	}
-
-	if confirm_tx != CommitmentType::RevokedCounterparty {
-		connect_blocks(&nodes[1], TEST_FINAL_CLTV - ANTI_REORG_DELAY + 1);
-		if confirm_tx == CommitmentType::LocalWithoutLastHTLC || confirm_tx == CommitmentType::LocalWithLastHTLC {
-			if !dust_htlcs {
-				handle_bump_events(&nodes[1], false, 1);
-			}
-		}
-	}
-
-	let bs_htlc_timeouts =
-		nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
-	if dust_htlcs || confirm_tx == CommitmentType::RevokedCounterparty {
-		assert_eq!(bs_htlc_timeouts.len(), 0);
-	} else {
-		assert_eq!(bs_htlc_timeouts.len(), 1);
-
-		// Now replay the timeouts on node B, which after 6 confirmations should fail the HTLCs via
-		// `MonitorUpdate`s
-		mine_transaction(&nodes[1], &bs_htlc_timeouts[0]);
-		connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
-	}
-
-	// Now simulate a restart where the B<->C ChannelMonitor has been persisted (i.e. because we
-	// just processed a new block) but the ChannelManager was not. This should be exceedingly rare
-	// given we have to be connecting a block at the right moment and not manage to get a
-	// ChannelManager persisted after it does a thing that should immediately precede persistence,
-	// but with async persist it is more common.
-	//
-	// We do this by wiping the `MonitorEvent`s from the monitors and then reloading with the
-	// latest state.
-	let mon_events = nodes[1].chain_monitor.chain_monitor.release_pending_monitor_events();
-	assert_eq!(mon_events.len(), 1);
-	assert_eq!(mon_events[0].2.len(), 3);
-
-	let node_ser = nodes[1].node.encode();
-	let mon_a_ser = get_monitor!(nodes[1], chan_a).encode();
-	let mon_b_ser = get_monitor!(nodes[1], chan_b).encode();
-	let mons = &[&mon_a_ser[..], &mon_b_ser[..]];
-	reload_node!(nodes[1], cfg, &node_ser, mons, persister, new_chain_mon, node_b_reload);
-
-	// After reload, once we process the `PaymentFailed` event, the sent HTLC will be marked
-	// handled so that we won't ever see the event again.
-	check_added_monitors(&nodes[1], 0);
-	let timeout_events = nodes[1].node.get_and_clear_pending_events();
-	check_added_monitors(&nodes[1], 1);
-	assert_eq!(timeout_events.len(), 3, "{timeout_events:?}");
-	for ev in timeout_events {
-		match ev {
-			Event::PaymentPathFailed { payment_hash, .. } => {
-				assert_eq!(payment_hash, hash_b);
-			},
-			Event::PaymentFailed { payment_hash, .. } => {
-				assert_eq!(payment_hash, Some(hash_b));
-			},
-			Event::HTLCHandlingFailed { prev_channel_ids, .. } => {
-				assert_eq!(prev_channel_ids[0], chan_a);
-			},
-			_ => panic!("Wrong event {ev:?}"),
-		}
-	}
-
-	nodes[0].node.peer_disconnected(nodes[1].node.get_our_node_id());
-
-	reconnect_nodes(ReconnectArgs::new(&nodes[0], &nodes[1]));
-
-	nodes[1].node.process_pending_htlc_forwards();
-	check_added_monitors(&nodes[1], 1);
-	let bs_fail = get_htlc_update_msgs(&nodes[1], &node_a_id);
-	nodes[0].node.handle_update_fail_htlc(node_b_id, &bs_fail.update_fail_htlcs[0]);
-	do_commitment_signed_dance(&nodes[0], &nodes[1], &bs_fail.commitment_signed, true, true);
-	expect_payment_failed!(nodes[0], hash_a, false);
-}
-
-#[test]
-fn test_lost_timeout_monitor_events() {
-	do_test_lost_timeout_monitor_events(CommitmentType::RevokedCounterparty, false, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::RevokedCounterparty, true, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::PreviousCounterparty, false, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::PreviousCounterparty, true, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::LatestCounterparty, false, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::LatestCounterparty, true, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::LocalWithoutLastHTLC, false, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::LocalWithoutLastHTLC, true, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::LocalWithLastHTLC, false, false);
-	do_test_lost_timeout_monitor_events(CommitmentType::LocalWithLastHTLC, true, false);
-
-	do_test_lost_timeout_monitor_events(CommitmentType::RevokedCounterparty, false, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::RevokedCounterparty, true, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::PreviousCounterparty, false, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::PreviousCounterparty, true, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::LatestCounterparty, false, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::LatestCounterparty, true, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::LocalWithoutLastHTLC, false, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::LocalWithoutLastHTLC, true, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::LocalWithLastHTLC, false, true);
-	do_test_lost_timeout_monitor_events(CommitmentType::LocalWithLastHTLC, true, true);
-}
-
 #[test]
 fn test_ladder_preimage_htlc_claims() {
 	// Tests that when we learn of a preimage via a monitor update we only claim HTLCs with the

### lightning/src/util/persist.rs
@@ -1543,17 +1543,23 @@ impl<
 		for (update_name, update_res) in MultiResultFuturePoller::new(update_futures).await {
 			let update = update_res?;
 			monitor
-				.update_monitor(&update, &self.broadcaster, &self.fee_estimator, &self.logger)
+				.update_monitor(
+					&update,
+					&self.broadcaster,
+					&self.fee_estimator,
+					&self.logger,
+					&self.entropy_source,
+				)
 				.map_err(|e| {
-				log_error!(
-					self.logger,
-					"Monitor update failed. monitor: {} update: {} reason: {:?}",
-					monitor_key,
-					update_name.as_str(),
-					e
-				);
-				io::Error::new(io::ErrorKind::Other, "Monitor update failed")
-			})?;
+					log_error!(
+						self.logger,
+						"Monitor update failed. monitor: {} update: {} reason: {:?}",
+						monitor_key,
+						update_name.as_str(),
+						e
+					);
+					io::Error::new(io::ErrorKind::Other, "Monitor update failed")
+				})?;
 		}
 		Ok(Some((best_block, monitor)))
 	}

### lightning/src/util/test_utils.rs
@@ -17,7 +17,7 @@ use crate::chain::chaininterface;
 #[cfg(any(test, feature = "_externalize_tests"))]
 use crate::chain::chaininterface::FEERATE_FLOOR_SATS_PER_KW;
 use crate::chain::chaininterface::{ConfirmationTarget, TransactionType};
-use crate::chain::chainmonitor::{ChainMonitor, Persist};
+use crate::chain::chainmonitor::{ChainMonitor, MonitorEventSource, Persist};
 use crate::chain::channelmonitor::{
 	ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, MonitorEvent,
 };
@@ -665,6 +665,7 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 		)
 		.unwrap()
 		.1;
+		new_monitor.copy_monitor_event_state(&monitor);
 		assert!(new_monitor == monitor);
 		self.latest_monitor_update_id
 			.lock()
@@ -726,6 +727,9 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 		// it so it doesn't leak into the rest of the test.
 		let failed_back = monitor.inner.lock().unwrap().failed_back_htlc_ids.clone();
 		new_monitor.inner.lock().unwrap().failed_back_htlc_ids = failed_back;
+		// The deserialized monitor will reset the monitor event state, so copy it from the live
+		// monitor before comparing.
+		new_monitor.copy_monitor_event_state(&monitor);
 		if let Some(chan_id) = self.expect_monitor_round_trip_fail.lock().unwrap().take() {
 			assert_eq!(chan_id, channel_id);
 			assert!(new_monitor != *monitor);
@@ -739,7 +743,7 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 
 	fn release_pending_monitor_events(
 		&self,
-	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
+	) -> Vec<(OutPoint, ChannelId, Vec<(u128, 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.)
@@ -750,6 +754,10 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 		}
 		return self.chain_monitor.release_pending_monitor_events();
 	}
+
+	fn ack_monitor_event(&self, source: MonitorEventSource) {
+		self.chain_monitor.ack_monitor_event(source);
+	}
 }
 
 #[cfg(any(test, feature = "_externalize_tests"))]