Pipe monitor event source HTLCForwardInfo -> holding cell htlcs

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
will simplify things - on restart instead of examining the set of HTLCs in
monitors we can simply replay all the pending MonitorEvents.

To ensure we can resolve HTLC monitor events for forward failures, we need to
pipe the event id through the HTLC failure pipeline. Here we pipe the monitor
event source from the manager to the channel, so it can be put in the channel's
holding cell.

In upcoming commits we will eventually get to the point of acking the monitor
event when the HTLC is irrevocably removed from the inbound edge via monitor
update.

Author: valentinewallace

lightning/src/ln/channel.rs

@@ -6457,7 +6457,9 @@ trait FailHTLCContents {
 	type Message: FailHTLCMessageName;
 	fn to_message(self, htlc_id: u64, channel_id: ChannelId) -> Self::Message;
 	fn to_inbound_htlc_state(self) -> InboundHTLCState;
-	fn to_htlc_update_awaiting_ack(self, htlc_id: u64) -> HTLCUpdateAwaitingACK;
+	fn to_htlc_update_awaiting_ack(
+		self, htlc_id: u64, monitor_event_source: Option<MonitorEventSource>,
+	) -> HTLCUpdateAwaitingACK;
 }
 impl FailHTLCContents for msgs::OnionErrorPacket {
 	type Message = msgs::UpdateFailHTLC;
@@ -6472,8 +6474,10 @@ impl FailHTLCContents for msgs::OnionErrorPacket {
 	fn to_inbound_htlc_state(self) -> InboundHTLCState {
 		InboundHTLCState::LocalRemoved(InboundHTLCRemovalReason::FailRelay(self))
 	}
-	fn to_htlc_update_awaiting_ack(self, htlc_id: u64) -> HTLCUpdateAwaitingACK {
-		HTLCUpdateAwaitingACK::FailHTLC { htlc_id, err_packet: self, monitor_event_source: None }
+	fn to_htlc_update_awaiting_ack(
+		self, htlc_id: u64, monitor_event_source: Option<MonitorEventSource>,
+	) -> HTLCUpdateAwaitingACK {
+		HTLCUpdateAwaitingACK::FailHTLC { htlc_id, err_packet: self, monitor_event_source }
 	}
 }
 impl FailHTLCContents for ([u8; 32], u16) {
@@ -6492,12 +6496,14 @@ impl FailHTLCContents for ([u8; 32], u16) {
 			failure_code: self.1,
 		})
 	}
-	fn to_htlc_update_awaiting_ack(self, htlc_id: u64) -> HTLCUpdateAwaitingACK {
+	fn to_htlc_update_awaiting_ack(
+		self, htlc_id: u64, monitor_event_source: Option<MonitorEventSource>,
+	) -> HTLCUpdateAwaitingACK {
 		HTLCUpdateAwaitingACK::FailMalformedHTLC {
 			htlc_id,
 			sha256_of_onion: self.0,
 			failure_code: self.1,
-			monitor_event_source: None,
+			monitor_event_source,
 		}
 	}
 }
@@ -7190,9 +7196,10 @@ where
 	/// Returns `Err` (always with [`ChannelError::Ignore`]) if the HTLC could not be failed (e.g.
 	/// if it was already resolved). Otherwise returns `Ok`.
 	pub fn queue_fail_htlc<L: Logger>(
-		&mut self, htlc_id_arg: u64, err_packet: msgs::OnionErrorPacket, logger: &L,
+		&mut self, htlc_id_arg: u64, err_packet: msgs::OnionErrorPacket,
+		monitor_event_source: Option<MonitorEventSource>, logger: &L,
 	) -> Result<(), ChannelError> {
-		self.fail_htlc(htlc_id_arg, err_packet, true, logger)
+		self.fail_htlc(htlc_id_arg, err_packet, true, monitor_event_source, logger)
 			.map(|msg_opt| assert!(msg_opt.is_none(), "We forced holding cell?"))
 	}
 
@@ -7201,18 +7208,25 @@ where
 	///
 	/// See [`Self::queue_fail_htlc`] for more info.
 	pub fn queue_fail_malformed_htlc<L: Logger>(
-		&mut self, htlc_id_arg: u64, failure_code: u16, sha256_of_onion: [u8; 32], logger: &L,
+		&mut self, htlc_id_arg: u64, failure_code: u16, sha256_of_onion: [u8; 32],
+		monitor_event_source: Option<MonitorEventSource>, logger: &L,
 	) -> Result<(), ChannelError> {
-		self.fail_htlc(htlc_id_arg, (sha256_of_onion, failure_code), true, logger)
-			.map(|msg_opt| assert!(msg_opt.is_none(), "We forced holding cell?"))
+		self.fail_htlc(
+			htlc_id_arg,
+			(sha256_of_onion, failure_code),
+			true,
+			monitor_event_source,
+			logger,
+		)
+		.map(|msg_opt| assert!(msg_opt.is_none(), "We forced holding cell?"))
 	}
 
 	/// Returns `Err` (always with [`ChannelError::Ignore`]) if the HTLC could not be failed (e.g.
 	/// if it was already resolved). Otherwise returns `Ok`.
 	#[rustfmt::skip]
 	fn fail_htlc<L: Logger, E: FailHTLCContents + Clone>(
 		&mut self, htlc_id_arg: u64, err_contents: E, mut force_holding_cell: bool,
-		logger: &L
+		monitor_event_source: Option<MonitorEventSource>, logger: &L
 	) -> Result<Option<E::Message>, ChannelError> {
 		if !matches!(self.context.channel_state, ChannelState::ChannelReady(_)) {
 			panic!("Was asked to fail an HTLC when channel was not in an operational state");
@@ -7267,7 +7281,7 @@ where
 				}
 			}
 			log_trace!(logger, "Placing failure for HTLC ID {} in holding cell.", htlc_id_arg);
-			self.context.holding_cell_htlc_updates.push(err_contents.to_htlc_update_awaiting_ack(htlc_id_arg));
+			self.context.holding_cell_htlc_updates.push(err_contents.to_htlc_update_awaiting_ack(htlc_id_arg, monitor_event_source));
 			return Ok(None);
 		}
 
@@ -8297,7 +8311,7 @@ where
 						None
 					},
 					&HTLCUpdateAwaitingACK::FailHTLC { htlc_id, ref err_packet, .. } => Some(
-						self.fail_htlc(htlc_id, err_packet.clone(), false, logger)
+						self.fail_htlc(htlc_id, err_packet.clone(), false, None, logger)
 							.map(|fail_msg_opt| fail_msg_opt.map(|_| ())),
 					),
 					&HTLCUpdateAwaitingACK::FailMalformedHTLC {
@@ -8306,8 +8320,14 @@ where
 						sha256_of_onion,
 						..
 					} => Some(
-						self.fail_htlc(htlc_id, (sha256_of_onion, failure_code), false, logger)
-							.map(|fail_msg_opt| fail_msg_opt.map(|_| ())),
+						self.fail_htlc(
+							htlc_id,
+							(sha256_of_onion, failure_code),
+							false,
+							None,
+							logger,
+						)
+						.map(|fail_msg_opt| fail_msg_opt.map(|_| ())),
 					),
 				};
 				if let Some(res) = fail_htlc_res {

lightning/src/ln/channelmanager.rs

@@ -8079,15 +8079,23 @@ impl<
 					}
 					None
 				},
-				HTLCForwardInfo::FailHTLC { htlc_id, ref err_packet, .. } => {
+				HTLCForwardInfo::FailHTLC { htlc_id, ref err_packet, monitor_event_source } => {
 					if let Some(chan) = peer_state
 						.channel_by_id
 						.get_mut(&forward_chan_id)
 						.and_then(Channel::as_funded_mut)
 					{
 						let logger = WithChannelContext::from(&self.logger, &chan.context, None);
 						log_trace!(logger, "Failing HTLC back to channel with short id {} (backward HTLC ID {}) after delay", short_chan_id, htlc_id);
-						Some((chan.queue_fail_htlc(htlc_id, err_packet.clone(), &&logger), htlc_id))
+						Some((
+							chan.queue_fail_htlc(
+								htlc_id,
+								err_packet.clone(),
+								monitor_event_source,
+								&&logger,
+							),
+							htlc_id,
+						))
 					} else {
 						self.forwarding_channel_not_found(
 							core::iter::once(forward_info).chain(draining_pending_forwards),
@@ -8103,7 +8111,7 @@ impl<
 					htlc_id,
 					failure_code,
 					sha256_of_onion,
-					..
+					monitor_event_source,
 				} => {
 					if let Some(chan) = peer_state
 						.channel_by_id
@@ -8116,6 +8124,7 @@ impl<
 							htlc_id,
 							failure_code,
 							sha256_of_onion,
+							monitor_event_source,
 							&&logger,
 						);
 						Some((res, htlc_id))