Prepare to provide new ReleasePaymentComplete monitor updates

`MonitorEvent`s aren't delivered to the `ChannelManager` in a
durable fasion - 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.

In previous commits we ensured that HTLC resolutions which came to
`ChannelManager` via a `MonitorEvent` were replayed on startup if
the `MonitorEvent` was lost. However, in cases where the
`ChannelManager` was so stale that it didn't have the payment state
for an HTLC at all, we only re-add it in cases where
`ChannelMonitor::get_pending_or_resolved_outbound_htlcs` includes
it.

Because constantly re-adding a payment state and then failing it
would generate lots of noise for users on startup (not to mention
risk of confusing stale payment events for the latest state of a
payment when the `PaymentId` has been reused to retry a payment).
Thus, `get_pending_or_resolved_outbound_htlcs` does not include
state for HTLCs which were resolved on chain with a preimage or
HTLCs which were resolved on chain with a timeout after
`ANTI_REORG_DELAY` confirmations.

This critera matches the critera for generating a `MonitorEvent`,
and works great under the assumption that `MonitorEvent`s are
reliably delivered. However, if they are not, and our
`ChannelManager` is lost or substantially old (or, in a future
where we do not persist `ChannelManager` at all), we will not end
up seeing payment resolution events for an HTLC.

Instead, we really want to tell our `ChannelMonitor`s when the
resolution of an HTLC is complete. Note that we don't particularly
care about non-payment HTLCs, as there is no re-hydration of state
to do there - `ChannelManager` load ignores forwarded HTLCs coming
back from `get_pending_or_resolved_outbound_htlcs` as there's
nothing to do - we always attempt to replay the success/failure and
figure out if it mattered based on whether there was still an HTLC
to claim/fail.

Here we prepare to generate the new
`ChannelMonitorUpdateStep::ReleasePaymentComplete` updates, adding
a new `PaymentCompleteUpdate` struct to track the new update before
we generate the `ChannelMonitorUpdate` and passing through to the
right places in `ChannelManager`.

The only cases where we want to generate the new update is after a
`PaymentSent` or `PaymentFailed` event when the event was the
result of a `MonitorEvent` or the equivalent read during startup.

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -1233,6 +1233,21 @@ impl_writeable_tlv_based_enum_upgradable!(MonitorUpdateCompletionAction,
 	},
 );
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct PaymentCompleteUpdate {
+	counterparty_node_id: PublicKey,
+	channel_funding_outpoint: OutPoint,
+	channel_id: ChannelId,
+	htlc_id: SentHTLCId,
+}
+
+impl_writeable_tlv_based!(PaymentCompleteUpdate, {
+	(1, channel_funding_outpoint, required),
+	(3, counterparty_node_id, required),
+	(5, channel_id, required),
+	(7, htlc_id, required),
+});
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub(crate) enum EventCompletionAction {
 	ReleaseRAAChannelMonitorUpdate {
@@ -3448,7 +3463,7 @@ macro_rules! handle_monitor_update_completion {
 		$self.finalize_claims(updates.finalized_claimed_htlcs);
 		for failure in updates.failed_htlcs.drain(..) {
 			let receiver = HTLCHandlingFailureType::Forward { node_id: Some(counterparty_node_id), channel_id };
-			$self.fail_htlc_backwards_internal(&failure.0, &failure.1, &failure.2, receiver);
+			$self.fail_htlc_backwards_internal(&failure.0, &failure.1, &failure.2, receiver, None);
 		}
 	} }
 }
@@ -4121,7 +4136,8 @@ where
 			let failure_reason = LocalHTLCFailureReason::ChannelClosed;
 			let reason = HTLCFailReason::from_failure_code(failure_reason);
 			let receiver = HTLCHandlingFailureType::Forward { node_id: Some(*counterparty_node_id), channel_id: *chan_id };
-			self.fail_htlc_backwards_internal(&htlc_source.0, &htlc_source.1, &reason, receiver);
+			let (source, hash) = htlc_source;
+			self.fail_htlc_backwards_internal(&source, &hash, &reason, receiver, None);
 		}
 
 		let _ = handle_error!(self, shutdown_result, *counterparty_node_id);
@@ -4255,7 +4271,7 @@ where
 			let failure_reason = LocalHTLCFailureReason::ChannelClosed;
 			let reason = HTLCFailReason::from_failure_code(failure_reason);
 			let receiver = HTLCHandlingFailureType::Forward { node_id: Some(counterparty_node_id), channel_id };
-			self.fail_htlc_backwards_internal(&source, &payment_hash, &reason, receiver);
+			self.fail_htlc_backwards_internal(&source, &payment_hash, &reason, receiver, None);
 		}
 		if let Some((_, funding_txo, _channel_id, monitor_update)) = shutdown_res.monitor_update {
 			debug_assert!(false, "This should have been handled in `locked_close_channel`");
@@ -6159,7 +6175,8 @@ where
 
 			let reason = HTLCFailReason::from_failure_code(LocalHTLCFailureReason::UnknownNextPeer);
 			let destination = HTLCHandlingFailureType::InvalidForward { requested_forward_scid: short_channel_id };
-			self.fail_htlc_backwards_internal(&htlc_source, &payment.forward_info.payment_hash, &reason, destination);
+			let hash = payment.forward_info.payment_hash;
+			self.fail_htlc_backwards_internal(&htlc_source, &hash, &reason, destination, None);
 		} else { unreachable!() } // Only `PendingHTLCRouting::Forward`s are intercepted
 
 		Ok(())
@@ -6448,6 +6465,7 @@ where
 				&payment_hash,
 				&failure_reason,
 				destination,
+				None,
 			);
 		}
 		self.forward_htlcs(&mut phantom_receives);
@@ -7657,7 +7675,7 @@ where
 				let failure_reason = LocalHTLCFailureReason::MPPTimeout;
 				let reason = HTLCFailReason::from_failure_code(failure_reason);
 				let receiver = HTLCHandlingFailureType::Receive { payment_hash: htlc_source.1 };
-				self.fail_htlc_backwards_internal(&source, &htlc_source.1, &reason, receiver);
+				self.fail_htlc_backwards_internal(&source, &htlc_source.1, &reason, receiver, None);
 			}
 
 			for (err, counterparty_node_id) in handle_errors {
@@ -7724,7 +7742,7 @@ where
 				let reason = self.get_htlc_fail_reason_from_failure_code(failure_code, &htlc);
 				let source = HTLCSource::PreviousHopData(htlc.prev_hop);
 				let receiver = HTLCHandlingFailureType::Receive { payment_hash: *payment_hash };
-				self.fail_htlc_backwards_internal(&source, &payment_hash, &reason, receiver);
+				self.fail_htlc_backwards_internal(&source, &payment_hash, &reason, receiver, None);
 			}
 		}
 	}
@@ -7817,7 +7835,7 @@ where
 				node_id: Some(counterparty_node_id.clone()),
 				channel_id,
 			};
-			self.fail_htlc_backwards_internal(&htlc_src, &payment_hash, &reason, receiver);
+			self.fail_htlc_backwards_internal(&htlc_src, &payment_hash, &reason, receiver, None);
 		}
 	}
 
@@ -7826,6 +7844,7 @@ where
 	fn fail_htlc_backwards_internal(
 		&self, source: &HTLCSource, payment_hash: &PaymentHash, onion_error: &HTLCFailReason,
 		failure_type: HTLCHandlingFailureType,
+		mut from_monitor_update_completion: Option<PaymentCompleteUpdate>,
 	) {
 		// Ensure that no peer state channel storage lock is held when calling this function.
 		// This ensures that future code doesn't introduce a lock-order requirement for
@@ -7857,7 +7876,17 @@ where
 					&self.secp_ctx,
 					&self.pending_events,
 					&self.logger,
+					&mut from_monitor_update_completion,
 				);
+				if let Some(update) = from_monitor_update_completion {
+					// If `fail_htlc` didn't `take` the post-event action, we should go ahead and
+					// complete it here as the failure was duplicative - we've already handled it.
+					// This should mostly only happen on startup, but it is possible to hit it in
+					// rare cases where a MonitorUpdate is replayed after restart because a
+					// ChannelMonitor wasn't persisted after it was applied (even though the
+					// ChannelManager was).
+					// TODO
+				}
 			},
 			HTLCSource::PreviousHopData(HTLCPreviousHopData {
 				ref short_channel_id,
@@ -7995,6 +8024,7 @@ where
 							&payment_hash,
 							&reason,
 							receiver,
+							None,
 						);
 					}
 					return;
@@ -8142,7 +8172,7 @@ where
 					err_data,
 				);
 				let receiver = HTLCHandlingFailureType::Receive { payment_hash };
-				self.fail_htlc_backwards_internal(&source, &payment_hash, &reason, receiver);
+				self.fail_htlc_backwards_internal(&source, &payment_hash, &reason, receiver, None);
 			}
 			self.claimable_payments.lock().unwrap().pending_claiming_payments.remove(&payment_hash);
 		}
@@ -8482,22 +8512,35 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 				debug_assert!(self.background_events_processed_since_startup.load(Ordering::Acquire),
 					"We don't support claim_htlc claims during startup - monitors may not be available yet");
 				debug_assert_eq!(next_channel_counterparty_node_id, path.hops[0].pubkey);
-				let ev_completion_action = EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
-					channel_funding_outpoint: next_channel_outpoint,
-					channel_id: next_channel_id,
-					counterparty_node_id: path.hops[0].pubkey,
-				};
+				let mut ev_completion_action =
+					Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
+						channel_funding_outpoint: next_channel_outpoint,
+						channel_id: next_channel_id,
+						counterparty_node_id: path.hops[0].pubkey,
+					});
 				self.pending_outbound_payments.claim_htlc(
 					payment_id,
 					payment_preimage,
 					bolt12_invoice,
 					session_priv,
 					path,
 					from_onchain,
-					ev_completion_action,
+					&mut ev_completion_action,
 					&self.pending_events,
 					&self.logger,
 				);
+				// If an event was generated, `claim_htlc` set `ev_completion_action` to None, if
+				// not, we should go ahead and run it now (as the claim was duplicative), at least
+				// if a PaymentClaimed event with the same action isn't already pending.
+				let have_action = if ev_completion_action.is_some() {
+					let pending_events = self.pending_events.lock().unwrap();
+					pending_events.iter().any(|(_, act)| *act == ev_completion_action)
+				} else {
+					false
+				};
+				if !have_action {
+					self.handle_post_event_actions(ev_completion_action);
+				}
 			},
 			HTLCSource::PreviousHopData(hop_data) => {
 				let prev_channel_id = hop_data.channel_id;
@@ -10030,7 +10073,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 				channel_id: msg.channel_id,
 			};
 			let reason = HTLCFailReason::from_failure_code(LocalHTLCFailureReason::ChannelClosed);
-			self.fail_htlc_backwards_internal(&htlc_source.0, &htlc_source.1, &reason, receiver);
+			let (source, hash) = htlc_source;
+			self.fail_htlc_backwards_internal(&source, &hash, &reason, receiver, None);
 		}
 
 		Ok(())
@@ -10508,6 +10552,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					&payment_hash,
 					&failure_reason,
 					destination,
+					None,
 				);
 			}
 
@@ -11101,6 +11146,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								&htlc_update.payment_hash,
 								&reason,
 								receiver,
+								None,
 							);
 						}
 					},
@@ -12623,8 +12669,8 @@ where
 		}
 	}
 
-	fn handle_post_event_actions(&self, actions: Vec<EventCompletionAction>) {
-		for action in actions {
+	fn handle_post_event_actions<I: IntoIterator<Item = EventCompletionAction>>(&self, actions: I) {
+		for action in actions.into_iter() {
 			match action {
 				EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
 					channel_funding_outpoint,
@@ -13479,7 +13525,7 @@ where
 		}
 
 		for (source, payment_hash, reason, destination) in timed_out_htlcs.drain(..) {
-			self.fail_htlc_backwards_internal(&source, &payment_hash, &reason, destination);
+			self.fail_htlc_backwards_internal(&source, &payment_hash, &reason, destination, None);
 		}
 	}
 
@@ -15684,7 +15730,7 @@ where
 					}
 					for (source, hash, cp_id, chan_id) in shutdown_result.dropped_outbound_htlcs {
 						let reason = LocalHTLCFailureReason::ChannelClosed;
-						failed_htlcs.push((source, hash, cp_id, chan_id, reason));
+						failed_htlcs.push((source, hash, cp_id, chan_id, reason, None));
 					}
 					channel_closures.push_back((
 						events::Event::ChannelClosed {
@@ -15728,6 +15774,7 @@ where
 								channel.context.get_counterparty_node_id(),
 								channel.context.channel_id(),
 								LocalHTLCFailureReason::ChannelClosed,
+								None,
 							));
 						}
 					}
@@ -16428,20 +16475,21 @@ where
 									// generating a `PaymentPathSuccessful` event but regenerating
 									// it and the `PaymentSent` on every restart until the
 									// `ChannelMonitor` is removed.
-									let compl_action =
+									let mut compl_action = Some(
 										EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
 											channel_funding_outpoint: monitor.get_funding_txo(),
 											channel_id: monitor.channel_id(),
 											counterparty_node_id: path.hops[0].pubkey,
-										};
+										},
+									);
 									pending_outbounds.claim_htlc(
 										payment_id,
 										preimage,
 										bolt12_invoice,
 										session_priv,
 										path,
 										false,
-										compl_action,
+										&mut compl_action,
 										&pending_events,
 										&&logger,
 									);
@@ -16456,12 +16504,20 @@ where
 							"Failing HTLC with payment hash {} as it was resolved on-chain.",
 							htlc.payment_hash
 						);
+						let completion_action = Some(PaymentCompleteUpdate {
+							counterparty_node_id: monitor.get_counterparty_node_id(),
+							channel_funding_outpoint: monitor.get_funding_txo(),
+							channel_id: monitor.channel_id(),
+							htlc_id: SentHTLCId::from_source(&htlc_source),
+						});
+
 						failed_htlcs.push((
 							htlc_source,
 							htlc.payment_hash,
 							monitor.get_counterparty_node_id(),
 							monitor.channel_id(),
 							LocalHTLCFailureReason::OnChainTimeout,
+							completion_action,
 						));
 					}
 				}
@@ -17147,11 +17203,13 @@ where
 		}
 
 		for htlc_source in failed_htlcs {
-			let (source, payment_hash, counterparty_id, channel_id, failure_reason) = htlc_source;
+			let (source, hash, counterparty_id, channel_id, failure_reason, ev_action) =
+				htlc_source;
 			let receiver =
 				HTLCHandlingFailureType::Forward { node_id: Some(counterparty_id), channel_id };
 			let reason = HTLCFailReason::from_failure_code(failure_reason);
-			channel_manager.fail_htlc_backwards_internal(&source, &payment_hash, &reason, receiver);
+			channel_manager
+				.fail_htlc_backwards_internal(&source, &hash, &reason, receiver, ev_action);
 		}
 
 		for (

lightning/src/ln/outbound_payment.rs

@@ -17,7 +17,9 @@ use lightning_invoice::Bolt11Invoice;
 use crate::blinded_path::{IntroductionNode, NodeIdLookUp};
 use crate::events::{self, PaidBolt12Invoice, PaymentFailureReason};
 use crate::ln::channel_state::ChannelDetails;
-use crate::ln::channelmanager::{EventCompletionAction, HTLCSource, PaymentId};
+use crate::ln::channelmanager::{
+	EventCompletionAction, HTLCSource, PaymentCompleteUpdate, PaymentId,
+};
 use crate::ln::onion_utils;
 use crate::ln::onion_utils::{DecodedOnionFailure, HTLCFailReason};
 use crate::offers::invoice::Bolt12Invoice;
@@ -2160,7 +2162,7 @@ impl OutboundPayments {
 	#[rustfmt::skip]
 	pub(super) fn claim_htlc<L: Deref>(
 		&self, payment_id: PaymentId, payment_preimage: PaymentPreimage, bolt12_invoice: Option<PaidBolt12Invoice>,
-		session_priv: SecretKey, path: Path, from_onchain: bool, ev_completion_action: EventCompletionAction,
+		session_priv: SecretKey, path: Path, from_onchain: bool, ev_completion_action: &mut Option<EventCompletionAction>,
 		pending_events: &Mutex<VecDeque<(events::Event, Option<EventCompletionAction>)>>,
 		logger: &L,
 	) where L::Target: Logger {
@@ -2181,7 +2183,7 @@ impl OutboundPayments {
 					amount_msat,
 					fee_paid_msat,
 					bolt12_invoice: bolt12_invoice,
-				}, Some(ev_completion_action.clone())));
+				}, ev_completion_action.take()));
 				payment.get_mut().mark_fulfilled();
 			}
 
@@ -2199,7 +2201,7 @@ impl OutboundPayments {
 						payment_hash,
 						path,
 						hold_times: Vec::new(),
-					}, Some(ev_completion_action)));
+					}, ev_completion_action.take()));
 				}
 			}
 		} else {
@@ -2328,7 +2330,7 @@ impl OutboundPayments {
 		path: &Path, session_priv: &SecretKey, payment_id: &PaymentId,
 		probing_cookie_secret: [u8; 32], secp_ctx: &Secp256k1<secp256k1::All>,
 		pending_events: &Mutex<VecDeque<(events::Event, Option<EventCompletionAction>)>>,
-		logger: &L,
+		logger: &L, completion_action: &mut Option<PaymentCompleteUpdate>,
 	) where
 		L::Target: Logger,
 	{
@@ -2479,6 +2481,7 @@ impl OutboundPayments {
 			}
 		};
 		let mut pending_events = pending_events.lock().unwrap();
+		// TODO: Handle completion_action
 		pending_events.push_back((path_failure, None));
 		if let Some(ev) = full_failure_ev {
 			pending_events.push_back((ev, None));