Support persistent monitor events

### lightning/src/chain/channelmonitor.rs
@@ -198,8 +198,9 @@ fn push_monitor_event<ES: EntropySource>(
 	pending_monitor_events.push((id, event));
 }
 
-/// An event to be processed by the ChannelManager.
-#[derive(Clone, PartialEq, Eq)]
+/// 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),
@@ -263,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>,
@@ -1310,6 +1311,12 @@ pub(crate) struct ChannelMonitorImpl<Signer: EcdsaChannelSigner> {
 	// 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<(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,
@@ -1775,7 +1782,12 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 			.map(|(_, ev)| ev)
 			.chain(holder_force_closed_compat.as_ref()),
 	));
-	let pending_mon_evs_with_ids = Some(Iterable(channel_monitor.pending_monitor_events.iter()));
+	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
@@ -1995,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,
 
@@ -2213,16 +2226,20 @@ 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`].
+	/// 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) {
-		// TODO: once events have ids, remove the corresponding event here
+	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`.
@@ -2231,15 +2248,23 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	/// original.
 	#[cfg(any(test, feature = "_test_utils"))]
 	pub fn copy_monitor_event_state(&self, other: &ChannelMonitor<Signer>) {
-		let pending = {
+		let (provided, pending) = {
 			let other_inner = other.inner.lock().unwrap();
-			other_inner.pending_monitor_events.clone()
+			(
+				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();
-		assert!(
-			self_inner.pending_monitor_events == pending,
+		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;
 	}
 
@@ -4707,9 +4732,16 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		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
 	}
 
@@ -6049,8 +6081,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 {
@@ -6541,13 +6573,15 @@ 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(
+					};
+					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,
@@ -7133,6 +7167,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 
 			payment_preimages,
 			pending_monitor_events,
+			provided_monitor_events: Vec::new(),
 			pending_events,
 			is_processing_pending_events: false,
 

### lightning/src/ln/monitor_tests.rs
@@ -3677,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