Generate 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 begin generating the new
`ChannelMonitorUpdateStep::ReleasePaymentComplete` updates,
updating functional tests for the new `ChannelMonitorUpdate`s where
required.

Author: TheBlueMatt

lightning/src/chain/channelmonitor.rs

@@ -4119,6 +4119,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		if updates.update_id == LEGACY_CLOSED_CHANNEL_UPDATE_ID || self.lockdown_from_offchain {
 			assert_eq!(updates.updates.len(), 1);
 			match updates.updates[0] {
+				ChannelMonitorUpdateStep::ReleasePaymentComplete { .. } => {},
 				ChannelMonitorUpdateStep::ChannelForceClosed { .. } => {},
 				// We should have already seen a `ChannelForceClosed` update if we're trying to
 				// provide a preimage at this point.

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -3870,6 +3870,7 @@ fn do_test_durable_preimages_on_closed_channel(
 	};
 	nodes[0].node.force_close_broadcasting_latest_txn(&chan_id_ab, &node_b_id, err_msg).unwrap();
 	check_closed_event(&nodes[0], 1, reason, false, &[node_b_id], 100000);
+	check_added_monitors(&nodes[0], 1);
 	let as_closing_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
 	assert_eq!(as_closing_tx.len(), 1);
 

lightning/src/ln/channelmanager.rs

@@ -1267,6 +1267,12 @@ pub(crate) enum EventCompletionAction {
 		channel_funding_outpoint: Option<OutPoint>,
 		channel_id: ChannelId,
 	},
+
+	/// When a payment's resolution is communicated to the downstream logic via
+	/// [`Event::PaymentSent`] or [`Event::PaymentFailed`] we may want to mark the payment as
+	/// fully-resolved in the [`ChannelMonitor`], which we do via this action.
+	/// Note that this action will be dropped on downgrade to LDK prior to 0.2!
+	ReleasePaymentCompleteChannelMonitorUpdate(PaymentCompleteUpdate),
 }
 impl_writeable_tlv_based_enum!(EventCompletionAction,
 	(0, ReleaseRAAChannelMonitorUpdate) => {
@@ -1279,6 +1285,7 @@ impl_writeable_tlv_based_enum!(EventCompletionAction,
 			ChannelId::v1_from_funding_outpoint(channel_funding_outpoint.unwrap())
 		})),
 	}
+	{1, ReleasePaymentCompleteChannelMonitorUpdate} => (),
 );
 
 /// The source argument which is passed to [`ChannelManager::claim_mpp_part`].
@@ -8013,7 +8020,17 @@ where
 					// 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
+					// For such cases, we also check that there's no existing pending event to
+					// complete this action already, which we let finish instead.
+					let action =
+						EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(update);
+					let have_action = {
+						let pending_events = self.pending_events.lock().unwrap();
+						pending_events.iter().any(|(_, act)| act.as_ref() == Some(&action))
+					};
+					if !have_action {
+						self.handle_post_event_actions([action]);
+					}
 				}
 			},
 			HTLCSource::PreviousHopData(HTLCPreviousHopData {
@@ -8642,19 +8659,34 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		next_user_channel_id: Option<u128>, attribution_data: Option<AttributionData>,
 		send_timestamp: Option<Duration>,
 	) {
+		debug_assert_eq!(
+			startup_replay,
+			!self.background_events_processed_since_startup.load(Ordering::Acquire)
+		);
+		let htlc_id = SentHTLCId::from_source(&source);
 		match source {
 			HTLCSource::OutboundRoute {
 				session_priv, payment_id, path, bolt12_invoice, ..
 			} => {
-				debug_assert!(self.background_events_processed_since_startup.load(Ordering::Acquire),
+				debug_assert!(!startup_replay,
 					"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 mut ev_completion_action =
+
+				let mut ev_completion_action = if from_onchain {
+					let release = PaymentCompleteUpdate {
+						counterparty_node_id: next_channel_counterparty_node_id,
+						channel_funding_outpoint: next_channel_outpoint,
+						channel_id: next_channel_id,
+						htlc_id,
+					};
+					Some(EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(release))
+				} else {
 					Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
 						channel_funding_outpoint: Some(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,
@@ -11372,12 +11404,18 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								channel_id,
 							};
 							let reason = HTLCFailReason::from_failure_code(failure_reason);
+							let completion_update = Some(PaymentCompleteUpdate {
+								counterparty_node_id,
+								channel_funding_outpoint: funding_outpoint,
+								channel_id,
+								htlc_id: SentHTLCId::from_source(&htlc_update.source),
+							});
 							self.fail_htlc_backwards_internal(
 								&htlc_update.source,
 								&htlc_update.payment_hash,
 								&reason,
 								receiver,
-								None,
+								completion_update,
 							);
 						}
 					},
@@ -12864,8 +12902,62 @@ where
 					channel_id,
 					counterparty_node_id,
 				} => {
+					let startup_complete =
+						self.background_events_processed_since_startup.load(Ordering::Acquire);
+					debug_assert!(startup_complete);
 					self.handle_monitor_update_release(counterparty_node_id, channel_id, None);
 				},
+				EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(
+					PaymentCompleteUpdate {
+						counterparty_node_id,
+						channel_funding_outpoint,
+						channel_id,
+						htlc_id,
+					},
+				) => {
+					let per_peer_state = self.per_peer_state.read().unwrap();
+					let mut peer_state = per_peer_state
+						.get(&counterparty_node_id)
+						.map(|state| state.lock().unwrap())
+						.expect("Channels originating a payment resolution must have peer state");
+					let update_id = peer_state
+						.closed_channel_monitor_update_ids
+						.get_mut(&channel_id)
+						.expect("Channels originating a payment resolution must have a monitor");
+					*update_id += 1;
+
+					let update = ChannelMonitorUpdate {
+						update_id: *update_id,
+						channel_id: Some(channel_id),
+						updates: vec![ChannelMonitorUpdateStep::ReleasePaymentComplete {
+							htlc: htlc_id,
+						}],
+					};
+
+					let during_startup =
+						!self.background_events_processed_since_startup.load(Ordering::Acquire);
+					if during_startup {
+						let event = BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+							counterparty_node_id,
+							funding_txo: channel_funding_outpoint,
+							channel_id,
+							update,
+						};
+						self.pending_background_events.lock().unwrap().push(event);
+					} else {
+						handle_new_monitor_update!(
+							self,
+							channel_funding_outpoint,
+							update,
+							peer_state,
+							peer_state,
+							per_peer_state,
+							counterparty_node_id,
+							channel_id,
+							POST_CHANNEL_CLOSE
+						);
+					}
+				},
 			}
 		}
 	}
@@ -16557,6 +16649,7 @@ where
 							monitor,
 							Some(htlc.payment_hash),
 						);
+						let htlc_id = SentHTLCId::from_source(&htlc_source);
 						match htlc_source {
 							HTLCSource::PreviousHopData(prev_hop_data) => {
 								let pending_forward_matches_htlc = |info: &PendingAddHTLCInfo| {
@@ -16614,6 +16707,15 @@ where
 							} => {
 								if let Some(preimage) = preimage_opt {
 									let pending_events = Mutex::new(pending_events_read);
+									let update = PaymentCompleteUpdate {
+										counterparty_node_id: monitor.get_counterparty_node_id(),
+										channel_funding_outpoint: monitor.get_funding_txo(),
+										channel_id: monitor.channel_id(),
+										htlc_id,
+									};
+									let mut compl_action = Some(
+										EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(update)
+									);
 									// Note that we set `from_onchain` to "false" here,
 									// deliberately keeping the pending payment around forever.
 									// Given it should only occur when we have a channel we're
@@ -16622,15 +16724,6 @@ where
 									// generating a `PaymentPathSuccessful` event but regenerating
 									// it and the `PaymentSent` on every restart until the
 									// `ChannelMonitor` is removed.
-									let mut compl_action = Some(
-										EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
-											channel_funding_outpoint: Some(
-												monitor.get_funding_txo(),
-											),
-											channel_id: monitor.channel_id(),
-											counterparty_node_id: path.hops[0].pubkey,
-										},
-									);
 									pending_outbounds.claim_htlc(
 										payment_id,
 										preimage,
@@ -16642,6 +16735,41 @@ where
 										&pending_events,
 										&&logger,
 									);
+									// If the completion action was not consumed, then there was no
+									// payment to claim, and we need to tell the `ChannelMonitor`
+									// we don't need to hear about the HTLC again, at least as long
+									// as the PaymentSent event isn't still sitting around in our
+									// event queue.
+									let have_action = if compl_action.is_some() {
+										let pending_events = pending_events.lock().unwrap();
+										pending_events.iter().any(|(_, act)| *act == compl_action)
+									} else {
+										false
+									};
+									if !have_action && compl_action.is_some() {
+										let mut peer_state = per_peer_state
+											.get(&counterparty_node_id)
+											.map(|state| state.lock().unwrap())
+											.expect("Channels originating a preimage must have peer state");
+										let update_id = peer_state
+											.closed_channel_monitor_update_ids
+											.get_mut(channel_id)
+											.expect("Channels originating a preimage must have a monitor");
+										*update_id += 1;
+
+										pending_background_events.push(BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+											counterparty_node_id: monitor.get_counterparty_node_id(),
+											funding_txo: monitor.get_funding_txo(),
+											channel_id: monitor.channel_id(),
+											update: ChannelMonitorUpdate {
+												update_id: *update_id,
+												channel_id: Some(monitor.channel_id()),
+												updates: vec![ChannelMonitorUpdateStep::ReleasePaymentComplete {
+													htlc: htlc_id,
+												}],
+											},
+										});
+									}
 									pending_events_read = pending_events.into_inner().unwrap();
 								}
 							},

lightning/src/ln/functional_test_utils.rs

@@ -2803,6 +2803,9 @@ pub fn expect_payment_sent<CM: AChannelManager, H: NodeHolder<CM = CM>>(
 	expected_fee_msat_opt: Option<Option<u64>>, expect_per_path_claims: bool,
 	expect_post_ev_mon_update: bool,
 ) -> (Option<PaidBolt12Invoice>, Vec<Event>) {
+	if expect_post_ev_mon_update {
+		check_added_monitors(node, 0);
+	}
 	let events = node.node().get_and_clear_pending_events();
 	let expected_payment_hash = PaymentHash(
 		bitcoin::hashes::sha256::Hash::hash(&expected_payment_preimage.0).to_byte_array(),
@@ -3052,6 +3055,7 @@ pub struct PaymentFailedConditions<'a> {
 	pub(crate) expected_blamed_chan_closed: Option<bool>,
 	pub(crate) expected_mpp_parts_remain: bool,
 	pub(crate) retry_expected: bool,
+	pub(crate) from_mon_update: bool,
 }
 
 impl<'a> PaymentFailedConditions<'a> {
@@ -3062,6 +3066,7 @@ impl<'a> PaymentFailedConditions<'a> {
 			expected_blamed_chan_closed: None,
 			expected_mpp_parts_remain: false,
 			retry_expected: false,
+			from_mon_update: false,
 		}
 	}
 	pub fn mpp_parts_remain(mut self) -> Self {
@@ -3086,6 +3091,10 @@ impl<'a> PaymentFailedConditions<'a> {
 		self.retry_expected = true;
 		self
 	}
+	pub fn from_mon_update(mut self) -> Self {
+		self.from_mon_update = true;
+		self
+	}
 }
 
 #[cfg(any(test, feature = "_externalize_tests"))]
@@ -3195,7 +3204,13 @@ pub fn expect_payment_failed_conditions<'a, 'b, 'c, 'd, 'e>(
 	node: &'a Node<'b, 'c, 'd>, expected_payment_hash: PaymentHash,
 	expected_payment_failed_permanently: bool, conditions: PaymentFailedConditions<'e>,
 ) {
+	if conditions.from_mon_update {
+		check_added_monitors(node, 0);
+	}
 	let events = node.node.get_and_clear_pending_events();
+	if conditions.from_mon_update {
+		check_added_monitors(node, 1);
+	}
 	expect_payment_failed_conditions_event(
 		events,
 		expected_payment_hash,