Add persistent_monitor_events flag to monitors/manager

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.

As a first step towards this, here we persist a flag in the ChannelManager and
ChannelMonitors indicating whether this new feature is enabled. It will be used
in upcoming commits to maintain compatibility and create an upgrade/downgrade
path between LDK versions.

Author: valentinewallace

lightning/src/chain/channelmonitor.rs

@@ -1281,6 +1281,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<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,
 
 	pub(super) pending_events: Vec<Event>,
 	pub(super) is_processing_pending_events: bool,
@@ -1732,8 +1738,12 @@ pub(crate) fn write_chanmon_internal<Signer: EcdsaChannelSigner, W: Writer>(
 		channel_monitor.pending_monitor_events.iter().chain(holder_force_closed_compat.as_ref()),
 	);
 
+	// 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(());
+
 	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, required), // Equivalent to required_vec because Iterable also writes as WithoutLength
 		(7, channel_monitor.funding_spend_seen, required),
@@ -1938,6 +1948,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 
 			payment_preimages: new_hash_map(),
 			pending_monitor_events: Vec::new(),
+			persistent_events_enabled: false,
 			pending_events: Vec::new(),
 			is_processing_pending_events: false,
 
@@ -6695,8 +6706,10 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 		let mut is_manual_broadcast = RequiredWrapper(None);
 		let mut funding_seen_onchain = RequiredWrapper(None);
 		let mut best_block_previous_blocks = None;
+		let mut persistent_events_enabled: Option<()> = None;
 		read_tlv_fields!(reader, {
 			(1, funding_spend_confirmed, option),
+			(2, persistent_events_enabled, option),
 			(3, htlcs_resolved_on_chain, optional_vec),
 			(5, pending_monitor_events, optional_vec),
 			(7, funding_spend_seen, option),
@@ -6723,6 +6736,12 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			best_block.previous_blocks = previous_blocks;
 		}
 
+		#[cfg(not(any(feature = "_test_utils", test)))]
+		if persistent_events_enabled.is_some() {
+			// This feature isn't supported yet so error if the writer expected it to be.
+			return Err(DecodeError::InvalidValue)
+		}
+
 		// Note that `payment_preimages_with_info` was added (and is always written) in LDK 0.1, so
 		// we can use it to determine if this monitor was last written by LDK 0.1 or later.
 		let written_by_0_1_or_later = payment_preimages_with_info.is_some();
@@ -6871,6 +6890,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 
 			payment_preimages,
 			pending_monitor_events: pending_monitor_events.unwrap(),
+			persistent_events_enabled: persistent_events_enabled.is_some(),
 			pending_events,
 			is_processing_pending_events: false,
 

lightning/src/ln/channelmanager.rs

@@ -2928,6 +2928,15 @@ pub struct ChannelManager<
 	/// [`ConfirmationTarget::MinAllowedNonAnchorChannelRemoteFee`] estimate.
 	last_days_feerates: Mutex<VecDeque<(u32, u32)>>,
 
+	/// When set, monitors will repeatedly provide an event back to the `ChannelManager` on restart
+	/// until the event is explicitly acknowledged as processed.
+	///
+	/// Allows us to reconstruct pending HTLC state by replaying monitor events on startup, rather
+	/// than from complexly polling and reconciling Channel{Monitor} APIs, as well as move the
+	/// responsibility of off-chain payment resolution from the Channel to the monitor. Will be
+	/// always set once support is complete.
+	persistent_monitor_events: bool,
+
 	#[cfg(test)]
 	pub(super) entropy_source: ES,
 	#[cfg(not(test))]
@@ -3658,6 +3667,8 @@ impl<
 			signer_provider,
 
 			logger,
+
+			persistent_monitor_events: false,
 		}
 	}
 
@@ -18102,6 +18113,9 @@ impl<
 			}
 		}
 
+		// Only write `persistent_events_enabled` if it's set to true, as it's an even TLV.
+		let persistent_monitor_events = self.persistent_monitor_events.then_some(());
+
 		write_tlv_fields!(writer, {
 			(1, pending_outbound_payments_no_retry, required),
 			(2, pending_intercepted_htlcs, option),
@@ -18114,6 +18128,7 @@ impl<
 			(9, htlc_purposes, required_vec),
 			(10, legacy_in_flight_monitor_updates, option),
 			(11, self.probing_cookie_secret, required),
+			(12, persistent_monitor_events, option),
 			(13, htlc_onion_fields, optional_vec),
 			(14, decode_update_add_htlcs_opt, option),
 			(15, self.inbound_payment_id_secret, required),
@@ -18213,6 +18228,7 @@ pub(super) struct ChannelManagerData<SP: SignerProvider> {
 	forward_htlcs_legacy: HashMap<u64, Vec<HTLCForwardInfo>>,
 	pending_intercepted_htlcs_legacy: HashMap<InterceptId, PendingAddHTLCInfo>,
 	decode_update_add_htlcs_legacy: HashMap<u64, Vec<msgs::UpdateAddHTLC>>,
+	persistent_monitor_events: bool,
 	// The `ChannelManager` version that was written.
 	version: u8,
 }
@@ -18399,6 +18415,7 @@ impl<'a, ES: EntropySource, SP: SignerProvider, L: Logger>
 		let mut peer_storage_dir: Option<Vec<(PublicKey, Vec<u8>)>> = None;
 		let mut async_receive_offer_cache: AsyncReceiveOfferCache = AsyncReceiveOfferCache::new();
 		let mut best_block_previous_blocks = None;
+		let mut persistent_monitor_events: Option<()> = None;
 		read_tlv_fields!(reader, {
 			(1, pending_outbound_payments_no_retry, option),
 			(2, pending_intercepted_htlcs_legacy, option),
@@ -18411,6 +18428,7 @@ impl<'a, ES: EntropySource, SP: SignerProvider, L: Logger>
 			(9, claimable_htlc_purposes, optional_vec),
 			(10, legacy_in_flight_monitor_updates, option),
 			(11, probing_cookie_secret, option),
+			(12, persistent_monitor_events, option),
 			(13, amountless_claimable_htlc_onion_fields, optional_vec),
 			(14, decode_update_add_htlcs_legacy, option),
 			(15, inbound_payment_id_secret, option),
@@ -18420,6 +18438,12 @@ impl<'a, ES: EntropySource, SP: SignerProvider, L: Logger>
 			(23, best_block_previous_blocks, option),
 		});
 
+		#[cfg(not(any(feature = "_test_utils", test)))]
+		if persistent_monitor_events.is_some() {
+			// This feature isn't supported yet so error if the writer expected it to be.
+			return Err(DecodeError::InvalidValue);
+		}
+
 		// Merge legacy pending_outbound_payments fields into a single HashMap.
 		// Priority: pending_outbound_payments (TLV 3) > pending_outbound_payments_no_retry (TLV 1)
 		//           > pending_outbound_payments_compat (non-TLV legacy)
@@ -18539,6 +18563,7 @@ impl<'a, ES: EntropySource, SP: SignerProvider, L: Logger>
 			peer_storage_dir: peer_storage_dir.unwrap_or_default(),
 			async_receive_offer_cache,
 			version,
+			persistent_monitor_events: persistent_monitor_events.is_some(),
 		})
 	}
 }
@@ -18839,6 +18864,7 @@ impl<
 			mut in_flight_monitor_updates,
 			peer_storage_dir,
 			async_receive_offer_cache,
+			persistent_monitor_events,
 			version: _version,
 		} = data;
 
@@ -20100,6 +20126,8 @@ impl<
 
 			logger: args.logger,
 			config: RwLock::new(args.config),
+
+			persistent_monitor_events,
 		};
 
 		let mut processed_claims: HashSet<Vec<MPPClaimHTLCSource>> = new_hash_set();