Persistent mon events for off-chain outbound claims

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.

In recent work, we added support for keeping monitor events around until they
are explicitly acked by the ChannelManager, but would always ack monitor events
immediately, which preserved the previous behavior and didn't break any tests.

Up until this point, we only generated HTLC monitor events when a payment was
claimed/failed on-chain.

In this commit, we start generating persistent monitor events whenever a payment
is claimed *off*-chain, specifically when new latest holder commitment data is
provided to the monitor.

For the purpose of making incremental progress on this feature, these events
will be a no-op and/or continue to be acked immediately except in the narrow
case of an off-chain outbound payment claim. HTLC forward claim monitor events
will be a no-op, and on-chain outbound payment claim events continue to be
acked immediately.

Off-chain outbound payment claims, however, now have monitor events generated
for them that will not be acked by the ChannelManager until the PaymentSent
event is processed by the user. This also allows us to stop blocking the RAA
monitor update that removes the preimage, because the purpose of that behavior
was to ensure the user got a PaymentSent event and the monitor event now serves
that purpose instead.

Author: valentinewallace

lightning/src/chain/channelmonitor.rs

@@ -2232,6 +2232,10 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		self.inner.lock().unwrap().persistent_events_enabled = enabled;
 	}
 
+	pub(crate) fn has_pending_event_for_htlc(&self, source: &HTLCSource) -> bool {
+		self.inner.lock().unwrap().has_pending_event_for_htlc(source)
+	}
+
 	/// Copies [`MonitorEvent`] state from `other` into `self`.
 	/// Used in tests to align transient runtime state before equality comparison after a
 	/// serialization round-trip.
@@ -3825,20 +3829,31 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		self.prev_holder_htlc_data = Some(htlc_data);
 
 		for (claimed_htlc_id, claimed_preimage) in claimed_htlcs {
-			#[cfg(debug_assertions)]
-			{
-				let cur_counterparty_htlcs = self
-					.funding
-					.counterparty_claimable_outpoints
-					.get(&self.funding.current_counterparty_commitment_txid.unwrap())
-					.unwrap();
-				assert!(cur_counterparty_htlcs.iter().any(|(_, source_opt)| {
+			let htlc_opt = self
+				.funding
+				.counterparty_claimable_outpoints
+				.get(&self.funding.current_counterparty_commitment_txid.unwrap())
+				.unwrap()
+				.iter()
+				.find_map(|(htlc, source_opt)| {
 					if let Some(source) = source_opt {
-						SentHTLCId::from_source(source) == *claimed_htlc_id
-					} else {
-						false
+						if SentHTLCId::from_source(source) == *claimed_htlc_id {
+							return Some((htlc, source));
+						}
 					}
-				}));
+					None
+				});
+			debug_assert!(htlc_opt.is_some());
+			if self.persistent_events_enabled {
+				if let Some((htlc, source)) = htlc_opt {
+					self.push_monitor_event(MonitorEvent::HTLCEvent(HTLCUpdate {
+						payment_hash: htlc.payment_hash,
+						payment_preimage: Some(*claimed_preimage),
+						source: *source.clone(),
+						htlc_value_satoshis: Some(htlc.amount_msat),
+						user_channel_id: self.user_channel_id,
+					}));
+				}
 			}
 			self.counterparty_fulfilled_htlcs.insert(*claimed_htlc_id, *claimed_preimage);
 		}
@@ -4656,6 +4671,16 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		self.pending_monitor_events.retain(|(id, _)| *id != event_id);
 	}
 
+	fn has_pending_event_for_htlc(&self, htlc: &HTLCSource) -> bool {
+		let htlc_id = SentHTLCId::from_source(htlc);
+		self.pending_monitor_events.iter().any(|(_, ev)| {
+			if let MonitorEvent::HTLCEvent(upd) = ev {
+				return htlc_id == SentHTLCId::from_source(&upd.source);
+			}
+			false
+		})
+	}
+
 	fn get_and_clear_pending_monitor_events(&mut self) -> Vec<(u64, MonitorEvent)> {
 		if self.persistent_events_enabled {
 			let mut ret = Vec::new();

lightning/src/ln/blinded_payment_tests.rs

@@ -854,7 +854,7 @@ fn do_blinded_intercept_payment(intercept_node_fails: bool) {
 	do_claim_payment_along_route(
 		ClaimAlongRouteArgs::new(&nodes[0], &[&[&nodes[1], &nodes[2]]], payment_preimage)
 	);
-	expect_payment_sent(&nodes[0], payment_preimage, Some(Some(1000)), true, true);
+	expect_payment_sent!(nodes[0], payment_preimage, Some(1000));
 }
 
 #[test]
@@ -1399,7 +1399,7 @@ fn conditionally_round_fwd_amt() {
 	let mut args = ClaimAlongRouteArgs::new(&nodes[0], &expected_route[..], payment_preimage)
 		.allow_1_msat_fee_overpay();
 	let expected_fee = pass_claimed_payment_along_route(args);
-	expect_payment_sent(&nodes[0], payment_preimage, Some(Some(expected_fee)), true, true);
+	expect_payment_sent!(nodes[0], payment_preimage, Some(expected_fee));
 }
 
 #[test]

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -2108,7 +2108,7 @@ fn monitor_update_claim_fail_no_response() {
 	let mut bs_updates = get_htlc_update_msgs(&nodes[1], &node_a_id);
 	nodes[0].node.handle_update_fulfill_htlc(node_b_id, bs_updates.update_fulfill_htlcs.remove(0));
 	do_commitment_signed_dance(&nodes[0], &nodes[1], &bs_updates.commitment_signed, false, false);
-	expect_payment_sent!(nodes[0], payment_preimage_1);
+	expect_payment_sent!(&nodes[0], payment_preimage_1);
 
 	claim_payment(&nodes[0], &[&nodes[1]], payment_preimage_2);
 }
@@ -3450,7 +3450,10 @@ fn do_test_blocked_chan_preimage_release(completion_mode: BlockedUpdateComplMode
 	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
 	let persister;
 	let new_chain_mon;
-	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let mut cfg = test_default_channel_config();
+	// If persistent_monitor_events is enabed, monitor updates will never be blocked.
+	cfg.override_persistent_monitor_events = Some(false);
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, Some(cfg.clone()), Some(cfg)]);
 	let nodes_1_reload;
 	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
 
@@ -3763,7 +3766,7 @@ fn do_test_inverted_mon_completion_order(
 	);
 
 	// Finally, check that the payment was, ultimately, seen as sent by node A.
-	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
+	expect_payment_sent!(&nodes[0], payment_preimage);
 }
 
 #[test]
@@ -4256,7 +4259,7 @@ fn do_test_glacial_peer_cant_hang(hold_chan_a: bool) {
 		nodes[0].node.handle_update_fulfill_htlc(node_b_id, update_fulfill);
 		let commitment = &a_update[0].commitment_signed;
 		do_commitment_signed_dance(&nodes[0], &nodes[1], commitment, false, false);
-		expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
+		expect_payment_sent!(nodes[0], payment_preimage);
 		expect_payment_forwarded!(nodes[1], nodes[0], nodes[2], Some(1000), false, false);
 
 		pass_along_path(
@@ -4936,6 +4939,7 @@ fn native_async_persist() {
 	let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
 	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
 	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+	let persistent_monitor_events = nodes[0].node.test_persistent_monitor_events_enabled();
 
 	let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1);
 
@@ -5081,7 +5085,16 @@ fn native_async_persist() {
 	assert_eq!(update_status, ChannelMonitorUpdateStatus::InProgress);
 
 	persist_futures.poll_futures();
-	assert_eq!(async_chain_monitor.release_pending_monitor_events().len(), 0);
+	let events = async_chain_monitor.release_pending_monitor_events();
+	if persistent_monitor_events {
+		// With persistent monitor events, the LatestHolderCommitmentTXInfo update containing
+		// claimed_htlcs generates an HTLCEvent with the preimage.
+		assert_eq!(events.len(), 1);
+		assert_eq!(events[0].2.len(), 1);
+		assert!(matches!(events[0].2[0].1, MonitorEvent::HTLCEvent(..)));
+	} else {
+		assert!(events.is_empty());
+	}
 
 	let pending_writes = kv_store.list_pending_async_writes(
 		CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
@@ -5223,7 +5236,7 @@ fn test_mpp_claim_to_holding_cell() {
 	let claims = vec![(b_claim_msgs, node_b_id), (c_claim_msgs, node_c_id)];
 	pass_claimed_payment_along_route_from_ev(250_000, claims, args);
 
-	expect_payment_sent(&nodes[0], preimage_1, None, true, true);
+	expect_payment_sent!(nodes[0], preimage_1);
 
 	expect_and_process_pending_htlcs(&nodes[3], false);
 	expect_payment_claimable!(nodes[3], paymnt_hash_2, payment_secret_2, 400_000);

lightning/src/ln/channelmanager.rs

@@ -10019,11 +10019,17 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					};
 					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,
-					})
+					if self.persistent_monitor_events {
+						monitor_event_id.map(|event_id| EventCompletionAction::AckMonitorEvent {
+							event_id: MonitorEventSource { channel_id: next_channel_id, event_id },
+						})
+					} else {
+						Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
+							channel_funding_outpoint: Some(next_channel_outpoint),
+							channel_id: next_channel_id,
+							counterparty_node_id: path.hops[0].pubkey,
+						})
+					}
 				};
 				let logger = WithContext::for_payment(
 					&self.logger,
@@ -10045,6 +10051,25 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					&self.pending_events,
 					&logger,
 				);
+
+				if matches!(
+					ev_completion_action,
+					Some(EventCompletionAction::AckMonitorEvent { .. })
+				) {
+					// If the `PaymentSent` for this redundant claim is still pending, add the event
+					// completion action here to ensure the `PaymentSent` will always be regenerated until it
+					// is processed by the user -- as long as the monitor event corresponding to this
+					// completion action is not acked, it will continue to be re-provided on startup.
+					let mut pending_events = self.pending_events.lock().unwrap();
+					for (ev, act_opt) in pending_events.iter_mut() {
+						let found_payment_sent = matches!(ev, Event::PaymentSent { payment_id: Some(id), .. } if *id == payment_id);
+						if found_payment_sent && act_opt.is_none() {
+							*act_opt = ev_completion_action.take();
+							break;
+						}
+					}
+				}
+
 				// 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.
@@ -12879,6 +12904,13 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		})
 	}
 
+	/// Returns `true` if `ChannelManager::persistent_monitor_events` is enabled. This flag will only
+	/// be set randomly in tests for now.
+	#[cfg(any(test, feature = "_test_utils"))]
+	pub fn test_persistent_monitor_events_enabled(&self) -> bool {
+		self.persistent_monitor_events
+	}
+
 	#[cfg(any(test, feature = "_test_utils"))]
 	pub(crate) fn test_raa_monitor_updates_held(
 		&self, counterparty_node_id: PublicKey, channel_id: ChannelId,
@@ -13585,22 +13617,29 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 										.channel_by_id
 										.contains_key(&channel_id)
 								});
-							// Claim the funds from the previous hop, if there is one. Because this is in response to a
-							// chain event, no attribution data is available.
-							self.claim_funds_internal(
-								htlc_update.source,
-								preimage,
-								htlc_update.htlc_value_satoshis.map(|v| v * 1000),
-								None,
-								from_onchain,
-								counterparty_node_id,
-								funding_outpoint,
-								channel_id,
-								htlc_update.user_channel_id,
-								None,
-								None,
-								Some(event_id),
-							);
+							let we_are_sender =
+								matches!(htlc_update.source, HTLCSource::OutboundRoute { .. });
+							if from_onchain | we_are_sender {
+								// Claim the funds from the previous hop, if there is one. Because this is in response to a
+								// chain event, no attribution data is available.
+								self.claim_funds_internal(
+									htlc_update.source,
+									preimage,
+									htlc_update.htlc_value_satoshis.map(|v| v * 1000),
+									None,
+									from_onchain,
+									counterparty_node_id,
+									funding_outpoint,
+									channel_id,
+									htlc_update.user_channel_id,
+									None,
+									None,
+									Some(event_id),
+								);
+							}
+							if from_onchain | !we_are_sender {
+								self.chain_monitor.ack_monitor_event(monitor_event_source);
+							}
 						} else {
 							log_trace!(logger, "Failing HTLC from our monitor");
 							let failure_reason = LocalHTLCFailureReason::OnChainTimeout;
@@ -13620,8 +13659,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								failure_type,
 								completion_update,
 							);
+							self.chain_monitor.ack_monitor_event(monitor_event_source);
 						}
-						self.chain_monitor.ack_monitor_event(monitor_event_source);
 					},
 					MonitorEvent::HolderForceClosed(_)
 					| MonitorEvent::HolderForceClosedWithInfo { .. } => {
@@ -19052,6 +19091,14 @@ impl<
 								break;
 							}
 						}
+						if persistent_monitor_events {
+							// This will not be necessary once we have persistent events for HTLC failures, we
+							// can delete this whole loop and wait to re-process the pending monitor events
+							// rather than failing them proactively below.
+							if monitor.has_pending_event_for_htlc(&channel_htlc_source) {
+								found_htlc = true;
+							}
+						}
 						if !found_htlc {
 							// If we have some HTLCs in the channel which are not present in the newer
 							// ChannelMonitor, they have been removed and should be failed back to

lightning/src/ln/functional_test_utils.rs

@@ -3074,7 +3074,7 @@ macro_rules! expect_payment_sent {
 			$expected_payment_preimage,
 			$expected_fee_msat_opt.map(|o| Some(o)),
 			$expect_paths,
-			true,
+			if $node.node.test_persistent_monitor_events_enabled() { false } else { true },
 		)
 	};
 }
@@ -4235,7 +4235,15 @@ pub fn claim_payment_along_route(
 		do_claim_payment_along_route(args) + expected_extra_total_fees_msat;
 
 	if !skip_last {
-		expect_payment_sent!(origin_node, payment_preimage, Some(expected_total_fee_msat))
+		let expect_post_ev_mon_update =
+			if origin_node.node.test_persistent_monitor_events_enabled() { false } else { true };
+		expect_payment_sent(
+			origin_node,
+			payment_preimage,
+			Some(Some(expected_total_fee_msat)),
+			true,
+			expect_post_ev_mon_update,
+		)
 	} else {
 		(None, Vec::new())
 	}

lightning/src/ln/functional_tests.rs

@@ -1358,6 +1358,10 @@ pub fn do_test_multiple_package_conflicts(p2a_anchor: bool) {
 	};
 	assert_eq!(updates.update_fulfill_htlcs.len(), 1);
 	nodes[0].node.handle_update_fulfill_htlc(node_b_id, updates.update_fulfill_htlcs.remove(0));
+	if nodes[0].node.test_persistent_monitor_events_enabled() {
+		// If persistent_monitor_events is enabled, the RAA monitor update is not blocked.
+		check_added_monitors(&nodes[0], 1);
+	}
 	do_commitment_signed_dance(&nodes[0], &nodes[1], &updates.commitment_signed, false, false);
 	expect_payment_sent!(nodes[0], preimage_2);
 
@@ -2672,7 +2676,10 @@ pub fn test_simple_peer_disconnect() {
 			_ => panic!("Unexpected event"),
 		}
 	}
-	check_added_monitors(&nodes[0], 1);
+	check_added_monitors(
+		&nodes[0],
+		if nodes[0].node.test_persistent_monitor_events_enabled() { 0 } else { 1 },
+	);
 
 	claim_payment(&nodes[0], &[&nodes[1], &nodes[2]], payment_preimage_4);
 	fail_payment(&nodes[0], &[&nodes[1], &nodes[2]], payment_hash_6);
@@ -4295,7 +4302,7 @@ pub fn test_duplicate_payment_hash_one_failure_one_success() {
 
 	nodes[0].node.handle_update_fulfill_htlc(node_b_id, updates.update_fulfill_htlcs.remove(0));
 	do_commitment_signed_dance(&nodes[0], &nodes[1], &updates.commitment_signed, false, false);
-	expect_payment_sent(&nodes[0], our_payment_preimage, None, true, true);
+	expect_payment_sent!(&nodes[0], our_payment_preimage);
 }
 
 #[xtest(feature = "_externalize_tests")]
@@ -8574,7 +8581,9 @@ pub fn test_inconsistent_mpp_params() {
 	pass_along_path(&nodes[0], path_b, real_amt, hash, Some(payment_secret), event, true, None);
 
 	do_claim_payment_along_route(ClaimAlongRouteArgs::new(&nodes[0], &[path_a, path_b], preimage));
-	expect_payment_sent(&nodes[0], preimage, Some(None), true, true);
+	let expect_post_ev_mon_update =
+		if nodes[0].node.test_persistent_monitor_events_enabled() { false } else { true };
+	expect_payment_sent(&nodes[0], preimage, Some(None), true, expect_post_ev_mon_update);
 }
 
 #[xtest(feature = "_externalize_tests")]
@@ -9932,7 +9941,10 @@ fn do_test_multi_post_event_actions(do_reload: bool) {
 	let chanmon_cfgs = create_chanmon_cfgs(3);
 	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
 	let (persister, chain_monitor);
-	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let mut cfg = test_default_channel_config();
+	// If persistent_monitor_events is enabled, RAAs will not be blocked on events.
+	cfg.override_persistent_monitor_events = Some(false);
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[Some(cfg), None, None]);
 	let node_a_reload;
 	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
 

lightning/src/ln/invoice_utils.rs

@@ -1344,7 +1344,7 @@ mod test {
 			&[&[&nodes[fwd_idx]]],
 			payment_preimage,
 		));
-		expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
+		expect_payment_sent!(&nodes[0], payment_preimage);
 	}
 
 	#[test]