Add monitor event ids

Currently, the resolution of HTLCs (and decisions on when HTLCs can be
forwarded) is the responsibility of Channel objects (a part of ChannelManager)
until the channel is closed, and then the ChannelMonitor thereafter. This leads
to some complexity around race conditions for HTLCs right around channel
closure. Additionally, there is lots of complexity reconstructing the state of
all HTLCs in the ChannelManager deserialization/loading logic.

Instead, we want to do all resolution in ChannelMonitors (in response to
ChannelMonitorUpdates) and pass them back to ChannelManager in the form of
MonitorEvents (similar to how HTLCs are resolved after channels are closed). In
order to have reliable resolution, we'll need to keep MonitorEvents around in
the ChannelMonitor until the ChannelManager has finished processing them.  This
simplify things - on restart instead of examining the set of HTLCs in monitors
we can simply replay all the pending MonitorEvents.

To allow the ChannelManager to ack specific monitor events once they are
resolved in upcoming commits, here we give each MonitorEvent a corresponding
unique id. It's implemented in such a way that we can delete legacy monitor
event code in the future when the new persistent monitor events flag is enabled
by default.

Author: valentinewallace

fuzz/src/chanmon_consistency.rs

@@ -365,7 +365,7 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 
 	fn release_pending_monitor_events(
 		&self,
-	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
+	) -> Vec<(OutPoint, ChannelId, Vec<(u64, MonitorEvent)>, PublicKey)> {
 		return self.chain_monitor.release_pending_monitor_events();
 	}
 

lightning/src/chain/chainmonitor.rs

@@ -1639,7 +1639,7 @@ where
 
 	fn release_pending_monitor_events(
 		&self,
-	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
+	) -> Vec<(OutPoint, ChannelId, Vec<(u64, MonitorEvent)>, PublicKey)> {
 		for (channel_id, update_id) in self.persister.get_and_clear_completed_updates() {
 			let _ = self.channel_monitor_updated(channel_id, update_id);
 		}

lightning/src/chain/channelmonitor.rs

@@ -183,8 +183,13 @@ impl Readable for ChannelMonitorUpdate {
 	}
 }
 
-fn push_monitor_event(pending_monitor_events: &mut Vec<MonitorEvent>, event: MonitorEvent) {
-	pending_monitor_events.push(event);
+fn push_monitor_event(
+	pending_monitor_events: &mut Vec<(u64, MonitorEvent)>, event: MonitorEvent,
+	next_monitor_event_id: &mut u64,
+) {
+	let id = *next_monitor_event_id;
+	*next_monitor_event_id += 1;
+	pending_monitor_events.push((id, event));
 }
 
 /// An event to be processed by the ChannelManager.
@@ -1282,13 +1287,14 @@ 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<(u64, MonitorEvent)>,
 	/// When set, monitor events are retained until explicitly acked rather than cleared on read.
 	///
 	/// Allows the ChannelManager to reconstruct pending HTLC state by replaying monitor events on
 	/// startup, and make the monitor responsible for both off- and on-chain payment resolution. Will
 	/// be always set once support for this feature is complete.
 	persistent_events_enabled: bool,
+	next_monitor_event_id: u64,
 
 	pub(super) pending_events: Vec<Event>,
 	pub(super) is_processing_pending_events: bool,
@@ -1669,32 +1675,38 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 		writer.write_all(&payment_preimage.0[..])?;
 	}
 
-	writer.write_all(
-		&(channel_monitor
-			.pending_monitor_events
-			.iter()
-			.filter(|ev| match ev {
-				MonitorEvent::HTLCEvent(_) => true,
-				MonitorEvent::HolderForceClosed(_) => true,
-				MonitorEvent::HolderForceClosedWithInfo { .. } => true,
-				_ => false,
-			})
-			.count() as u64)
-			.to_be_bytes(),
-	)?;
-	for event in channel_monitor.pending_monitor_events.iter() {
-		match event {
-			MonitorEvent::HTLCEvent(upd) => {
-				0u8.write(writer)?;
-				upd.write(writer)?;
-			},
-			MonitorEvent::HolderForceClosed(_) => 1u8.write(writer)?,
-			// `HolderForceClosedWithInfo` replaced `HolderForceClosed` in v0.0.122. To keep
-			// backwards compatibility, we write a `HolderForceClosed` event along with the
-			// `HolderForceClosedWithInfo` event. This is deduplicated in the reader.
-			MonitorEvent::HolderForceClosedWithInfo { .. } => 1u8.write(writer)?,
-			_ => {}, // Covered in the TLV writes below
+	if !channel_monitor.persistent_events_enabled {
+		writer.write_all(
+			&(channel_monitor
+				.pending_monitor_events
+				.iter()
+				.filter(|(_, ev)| match ev {
+					MonitorEvent::HTLCEvent(_) => true,
+					MonitorEvent::HolderForceClosed(_) => true,
+					MonitorEvent::HolderForceClosedWithInfo { .. } => true,
+					_ => false,
+				})
+				.count() as u64)
+				.to_be_bytes(),
+		)?;
+		for (_, event) in channel_monitor.pending_monitor_events.iter() {
+			match event {
+				MonitorEvent::HTLCEvent(upd) => {
+					0u8.write(writer)?;
+					upd.write(writer)?;
+				},
+				MonitorEvent::HolderForceClosed(_) => 1u8.write(writer)?,
+				// `HolderForceClosedWithInfo` replaced `HolderForceClosed` in v0.0.122. To keep
+				// backwards compatibility, we write a `HolderForceClosed` event along with the
+				// `HolderForceClosedWithInfo` event. This is deduplicated in the reader.
+				MonitorEvent::HolderForceClosedWithInfo { .. } => 1u8.write(writer)?,
+				_ => {}, // Covered in the TLV writes below
+			}
 		}
+	} else {
+		// If `persistent_events_enabled` is set, we'll write the events with their event ids in the
+		// TLV section below.
+		writer.write_all(&(0u64).to_be_bytes())?;
 	}
 
 	writer.write_all(&(channel_monitor.pending_events.len() as u64).to_be_bytes())?;
@@ -1729,25 +1741,40 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 
 	// 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 = Iterable(
-		channel_monitor.pending_monitor_events.iter().chain(holder_force_closed_compat.as_ref()),
-	);
+	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 = if !channel_monitor.persistent_events_enabled {
+		Some(Iterable(
+			channel_monitor
+				.pending_monitor_events
+				.iter()
+				.map(|(_, ev)| ev)
+				.chain(holder_force_closed_compat.as_ref()),
+		))
+	} else {
+		None
+	};
 
 	// Only write `persistent_events_enabled` if it's set to true, as it's an even TLV.
 	let persistent_events_enabled = channel_monitor.persistent_events_enabled.then_some(());
+	let pending_mon_evs_with_ids = if persistent_events_enabled.is_some() {
+		Some(Iterable(channel_monitor.pending_monitor_events.iter()))
+	} else {
+		None
+	};
 
 	write_tlv_fields!(writer, {
 		(1, channel_monitor.funding_spend_confirmed, option),
 		(2, persistent_events_enabled, option),
 		(3, channel_monitor.htlcs_resolved_on_chain, required_vec),
-		(5, pending_monitor_events_legacy, required), // Equivalent to required_vec because Iterable also writes as WithoutLength
+		(4, pending_mon_evs_with_ids, option),
+		(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),
@@ -1767,6 +1794,7 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 		(35, channel_monitor.is_manual_broadcast, required),
 		(37, channel_monitor.funding_seen_onchain, required),
 		(39, channel_monitor.best_block.previous_blocks, required),
+		(41, channel_monitor.next_monitor_event_id, required),
 	});
 
 	Ok(())
@@ -1951,6 +1979,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 			payment_preimages: new_hash_map(),
 			pending_monitor_events: Vec::new(),
 			persistent_events_enabled: false,
+			next_monitor_event_id: 0,
 			pending_events: Vec::new(),
 			is_processing_pending_events: false,
 
@@ -2164,7 +2193,7 @@ 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> {
+	pub fn get_and_clear_pending_monitor_events(&self) -> Vec<(u64, MonitorEvent)> {
 		self.inner.lock().unwrap().get_and_clear_pending_monitor_events()
 	}
 
@@ -2178,6 +2207,20 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		// TODO: once events have ids, remove the corresponding event here
 	}
 
+	/// Copies [`MonitorEvent`] state from `other` into `self`.
+	/// Used in tests to align transient runtime state before equality comparison after a
+	/// serialization round-trip.
+	#[cfg(any(test, feature = "_test_utils"))]
+	pub fn copy_monitor_event_state(&self, other: &ChannelMonitor<Signer>) {
+		let (pending, next_id) = {
+			let other_inner = other.inner.lock().unwrap();
+			(other_inner.pending_monitor_events.clone(), other_inner.next_monitor_event_id)
+		};
+		let mut self_inner = self.inner.lock().unwrap();
+		self_inner.pending_monitor_events = pending;
+		self_inner.next_monitor_event_id = next_id;
+	}
+
 	/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity.
 	///
 	/// For channels featuring anchor outputs, this method will also process [`BumpTransaction`]
@@ -3903,7 +3946,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 				outpoint: funding_outpoint,
 				channel_id: self.channel_id,
 			};
-			push_monitor_event(&mut self.pending_monitor_events, event);
+			push_monitor_event(&mut self.pending_monitor_events, event, &mut self.next_monitor_event_id);
 		}
 
 		// Although we aren't signing the transaction directly here, the transaction will be signed
@@ -4494,12 +4537,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,
+					}),
+					&mut self.next_monitor_event_id,
+				);
 				self.htlcs_resolved_on_chain.push(IrrevocablyResolvedHTLC {
 					commitment_tx_output_idx: None,
 					resolving_txid: Some(confirmed_txid),
@@ -4565,10 +4612,14 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 
 	fn push_monitor_event(&mut self, event: MonitorEvent) {
-		push_monitor_event(&mut self.pending_monitor_events, event);
+		push_monitor_event(
+			&mut self.pending_monitor_events,
+			event,
+			&mut self.next_monitor_event_id,
+		);
 	}
 
-	fn get_and_clear_pending_monitor_events(&mut self) -> Vec<MonitorEvent> {
+	fn get_and_clear_pending_monitor_events(&mut self) -> Vec<(u64, MonitorEvent)> {
 		let mut ret = Vec::new();
 		mem::swap(&mut ret, &mut self.pending_monitor_events);
 		ret
@@ -5899,7 +5950,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					continue;
 				}
 				let duplicate_event = self.pending_monitor_events.iter().any(
-					|update| if let &MonitorEvent::HTLCEvent(ref upd) = update {
+					|(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update {
 						upd.source == *source
 					} else { false });
 				if duplicate_event {
@@ -5917,7 +5968,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 						payment_preimage: None,
 						payment_hash: htlc.payment_hash,
 						htlc_value_satoshis: Some(htlc.amount_msat / 1000),
-					}));
+					}), &mut self.next_monitor_event_id);
 				}
 			}
 		}
@@ -6316,7 +6367,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 			if let Some((source, payment_hash, amount_msat)) = payment_data {
 				if accepted_preimage_claim {
 					if !self.pending_monitor_events.iter().any(
-						|update| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) {
+						|(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) {
 						self.onchain_events_awaiting_threshold_conf.push(OnchainEventEntry {
 							txid: tx.compute_txid(),
 							height,
@@ -6334,11 +6385,11 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 							payment_preimage: Some(payment_preimage),
 							payment_hash,
 							htlc_value_satoshis: Some(amount_msat / 1000),
-						}));
+						}), &mut self.next_monitor_event_id);
 					}
 				} else if offered_preimage_claim {
 					if !self.pending_monitor_events.iter().any(
-						|update| if let &MonitorEvent::HTLCEvent(ref upd) = update {
+						|(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update {
 							upd.source == source
 						} else { false }) {
 						self.onchain_events_awaiting_threshold_conf.push(OnchainEventEntry {
@@ -6358,7 +6409,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 							payment_preimage: Some(payment_preimage),
 							payment_hash,
 							htlc_value_satoshis: Some(amount_msat / 1000),
-						}));
+						}), &mut self.next_monitor_event_id);
 					}
 				} else {
 					self.onchain_events_awaiting_threshold_conf.retain(|ref entry| {
@@ -6723,10 +6774,13 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 		let mut funding_seen_onchain = RequiredWrapper(None);
 		let mut best_block_previous_blocks = None;
 		let mut persistent_events_enabled: Option<()> = None;
+		let mut next_monitor_event_id: Option<u64> = None;
+		let mut pending_mon_evs_with_ids: Option<Vec<ReadableIdMonitorEvent>> = None;
 		read_tlv_fields!(reader, {
 			(1, funding_spend_confirmed, option),
 			(2, persistent_events_enabled, option),
 			(3, htlcs_resolved_on_chain, optional_vec),
+			(4, pending_mon_evs_with_ids, optional_vec),
 			(5, pending_monitor_events_legacy, optional_vec),
 			(7, funding_spend_seen, option),
 			(9, counterparty_node_id, option),
@@ -6747,6 +6801,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			(35, is_manual_broadcast, (default_value, false)),
 			(37, funding_seen_onchain, (default_value, true)),
 			(39, best_block_previous_blocks, option), // Added and always set in 0.3
+			(41, next_monitor_event_id, option),
 		});
 		if let Some(previous_blocks) = best_block_previous_blocks {
 			best_block.previous_blocks = previous_blocks;
@@ -6788,6 +6843,22 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			}
 		}
 
+		// If persistent events are enabled, use the events with their persisted IDs from TLV 4.
+		// Otherwise, use the legacy events from TLV 5 and assign sequential IDs.
+		let (next_monitor_event_id, pending_monitor_events): (u64, Vec<(u64, MonitorEvent)>) =
+			if persistent_events_enabled.is_some() {
+				let evs = pending_mon_evs_with_ids.unwrap_or_default()
+					.into_iter().map(|ev| (ev.0, ev.1)).collect();
+				(next_monitor_event_id.unwrap_or(0), evs)
+			} else if let Some(events) = pending_monitor_events_legacy {
+				let next_id = next_monitor_event_id.unwrap_or(events.len() as u64);
+				let evs = events.into_iter().enumerate()
+					.map(|(i, ev)| (i as u64, ev)).collect();
+				(next_id, evs)
+			} else {
+				(next_monitor_event_id.unwrap_or(0), Vec::new())
+			};
+
 		let channel_parameters = channel_parameters.unwrap_or_else(|| {
 			onchain_tx_handler.channel_parameters().clone()
 		});
@@ -6905,8 +6976,9 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			current_holder_commitment_number,
 
 			payment_preimages,
-			pending_monitor_events: pending_monitor_events_legacy.unwrap(),
+			pending_monitor_events,
 			persistent_events_enabled: persistent_events_enabled.is_some(),
+			next_monitor_event_id,
 			pending_events,
 			is_processing_pending_events: false,
 
@@ -6955,6 +7027,22 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 	}
 }
 
+/// Deserialization wrapper for reading a `(u64, MonitorEvent)`.
+/// Necessary because we can't deserialize a (Readable, MaybeReadable) tuple due to trait
+/// conflicts.
+struct ReadableIdMonitorEvent(u64, MonitorEvent);
+
+impl MaybeReadable for ReadableIdMonitorEvent {
+	fn read<R: io::Read>(reader: &mut R) -> Result<Option<Self>, DecodeError> {
+		let id: u64 = 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, wrap_signer: impl FnOnce(crate::sign::InMemorySigner) -> S,