Add chain::Watch ack_monitor_event API

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.

Here we add an as-yet-unused API to chain::Watch to allow the ChannelManager to
tell the a ChannelMonitor that a MonitorEvent has been irrevocably processed
and can be deleted.

Author: valentinewallace

fuzz/src/chanmon_consistency.rs

@@ -41,6 +41,7 @@ use lightning::chain;
 use lightning::chain::chaininterface::{
 	BroadcasterInterface, ConfirmationTarget, FeeEstimator, TransactionType,
 };
+use lightning::chain::chainmonitor::MonitorEventSource;
 use lightning::chain::channelmonitor::{ChannelMonitor, MonitorEvent};
 use lightning::chain::transaction::OutPoint;
 use lightning::chain::{
@@ -367,6 +368,10 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
 		return self.chain_monitor.release_pending_monitor_events();
 	}
+
+	fn ack_monitor_event(&self, source: MonitorEventSource) {
+		self.chain_monitor.ack_monitor_event(source);
+	}
 }
 
 struct KeyProvider {

lightning/src/chain/chainmonitor.rs

@@ -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 event ID assigned by the [`ChannelMonitor`].
+	pub event_id: u64,
+	/// The channel from which the [`MonitorEvent`] originated.
+	pub channel_id: ChannelId,
+}
+
+impl_writeable_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.
@@ -1646,6 +1661,15 @@ where
 		}
 		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 {
+			debug_assert!(false, "Ack'd monitor events should always have a corresponding monitor");
+		}
+	}
 }
 
 impl<

lightning/src/chain/channelmonitor.rs

@@ -2172,6 +2172,12 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		self.inner.lock().unwrap().push_monitor_event(event);
 	}
 
+	/// 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: u64) {
+		// TODO: once events have ids, remove the corresponding event here
+	}
+
 	/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity.
 	///
 	/// For channels featuring anchor outputs, this method will also process [`BumpTransaction`]

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,
 };
@@ -425,6 +426,15 @@ pub trait Watch<ChannelSigner: EcdsaChannelSigner> {
 	fn release_pending_monitor_events(
 		&self,
 	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)>;
+
+	/// Acknowledges and removes a [`MonitorEvent`] previously returned by
+	/// [`Watch::release_pending_monitor_events`] by its event ID.
+	///
+	/// Once 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>>
@@ -447,6 +457,10 @@ impl<ChannelSigner: EcdsaChannelSigner, T: Watch<ChannelSigner> + ?Sized, W: Der
 	) -> Vec<(OutPoint, ChannelId, Vec<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/util/test_utils.rs

@@ -15,7 +15,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,
 };
@@ -734,6 +734,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"))]