Persistent monitor events for HTLC forward 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.

Building on the work of prior commits, here we stop immediately acking monitor
events for HTLC claims for forwarded HTLCs, and instead ack the monitor event
when the forward is fully resolved and we don't want the monitor event to be
re-provided back to us on startup.

If the inbound edge channel is on-chain, we'll ack the event when the inbound
edge preimage monitor update completes and the user processes a
PaymentForwarded event. If the inbound edge is open, we'll ack the event when
the inbound HTLC is fully removed via revoke_and_ack (this ensures we'll
remember to resolve the htlc off-chain even if we lose the holding cell).

Author: valentinewallace

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -3633,14 +3633,23 @@ impl TrustedChannelFeatures {
 struct ClaimCompletionActionParams {
 	definitely_duplicate: bool,
 	inbound_htlc_value_msat: Option<u64>,
+	inbound_edge_closed: bool,
 }
 
 impl ClaimCompletionActionParams {
 	fn new_claim(inbound_htlc_value_msat: u64) -> Self {
-		Self { definitely_duplicate: false, inbound_htlc_value_msat: Some(inbound_htlc_value_msat) }
+		Self {
+			definitely_duplicate: false,
+			inbound_htlc_value_msat: Some(inbound_htlc_value_msat),
+			inbound_edge_closed: false,
+		}
 	}
 	fn duplicate_claim() -> Self {
-		Self { definitely_duplicate: true, inbound_htlc_value_msat: None }
+		Self {
+			definitely_duplicate: true,
+			inbound_htlc_value_msat: None,
+			inbound_edge_closed: false,
+		}
 	}
 }
 
@@ -9649,16 +9658,56 @@ impl<
 			monitor_event_id
 				.map(|event_id| MonitorEventSource { event_id, channel_id: next_channel_id }),
 			|claim_completion_action_params| {
-				let ClaimCompletionActionParams { definitely_duplicate, inbound_htlc_value_msat } =
-					claim_completion_action_params;
+				let ClaimCompletionActionParams {
+					definitely_duplicate,
+					inbound_htlc_value_msat,
+					inbound_edge_closed,
+				} = claim_completion_action_params;
 				let chan_to_release = EventUnblockedChannel {
 					counterparty_node_id: next_channel_counterparty_node_id,
 					funding_txo: next_channel_outpoint,
 					channel_id: next_channel_id,
 					blocking_action: completed_blocker,
 				};
 
-				if definitely_duplicate && startup_replay {
+				if self.persistent_monitor_events {
+					let monitor_event_source = monitor_event_id.map(|event_id| {
+						MonitorEventSource { event_id, channel_id: next_channel_id }
+					});
+					// If persistent_monitor_events is enabled, then we'll get a MonitorEvent for this HTLC
+					// claim re-provided to us until we explicitly ack it.
+					// * If the inbound edge is closed, then we can ack it when we know the preimage is
+					// durably persisted there + the user has processed a `PaymentForwarded` event
+					// * If the inbound edge is open, then we'll ack the monitor event when HTLC has been
+					// irrevocably removed via revoke_and_ack. This prevents forgetting to claim the HTLC
+					// backwards if we lose the off-chain HTLC from the holding cell after a restart.
+					if definitely_duplicate {
+						if inbound_edge_closed {
+							if let Some(id) = monitor_event_source {
+								self.chain_monitor.ack_monitor_event(id);
+							}
+						}
+						(None, None)
+					} else if let Some(event) =
+						make_payment_forwarded_event(inbound_htlc_value_msat)
+					{
+						let preimage_update_action =
+							MonitorUpdateCompletionAction::EmitForwardEvent {
+								event,
+								post_event_ackable_monitor_event: inbound_edge_closed
+									.then_some(monitor_event_source)
+									.flatten(),
+							};
+						(Some(preimage_update_action), None)
+					} else if inbound_edge_closed {
+						let preimage_update_action = monitor_event_source.map(|src| {
+							MonitorUpdateCompletionAction::AckMonitorEvents { event_ids: vec![src] }
+						});
+						(preimage_update_action, None)
+					} else {
+						(None, None)
+					}
+				} else if definitely_duplicate && startup_replay {
 					// On startup we may get redundant claims which are related to
 					// monitor updates still in flight. In that case, we shouldn't
 					// immediately free, but instead let that monitor update complete
@@ -9991,6 +10040,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		let (action_opt, raa_blocker_opt) = completion_action(ClaimCompletionActionParams {
 			definitely_duplicate: false,
 			inbound_htlc_value_msat: None,
+			inbound_edge_closed: true,
 		});
 
 		if let Some(raa_blocker) = raa_blocker_opt {
@@ -12712,23 +12762,32 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 							chan.update_fulfill_htlc(&msg),
 							chan_entry
 						);
-						let prev_hops = match &res.0 {
-							HTLCSource::PreviousHopData(prev_hop) => vec![prev_hop],
-							HTLCSource::TrampolineForward { previous_hop_data, .. } => {
-								previous_hop_data.iter().collect()
-							},
-							_ => vec![],
-						};
-						let logger = WithChannelContext::from(&self.logger, &chan.context, None);
-						for prev_hop in prev_hops {
-							log_trace!(logger,
-								"Holding the next revoke_and_ack until the preimage is durably persisted in the inbound edge's ChannelMonitor",
-							);
-							peer_state
-								.actions_blocking_raa_monitor_updates
-								.entry(msg.channel_id)
-								.or_insert_with(Vec::new)
-								.push(RAAMonitorUpdateBlockingAction::from_prev_hop_data(prev_hop));
+						// If persistent_monitor_events is enabled, we don't need to block preimage-removing
+						// monitor updates because we'll get the preimage from monitor events (that are
+						// guaranteed to be re-provided until they are explicitly acked) rather than from
+						// polling the monitor's internal state.
+						if !self.persistent_monitor_events {
+							let prev_hops = match &res.0 {
+								HTLCSource::PreviousHopData(prev_hop) => vec![prev_hop],
+								HTLCSource::TrampolineForward { previous_hop_data, .. } => {
+									previous_hop_data.iter().collect()
+								},
+								_ => vec![],
+							};
+							let logger =
+								WithChannelContext::from(&self.logger, &chan.context, None);
+							for prev_hop in prev_hops {
+								log_trace!(logger,
+									"Holding the next revoke_and_ack until the preimage is durably persisted in the inbound edge's ChannelMonitor",
+								);
+								peer_state
+									.actions_blocking_raa_monitor_updates
+									.entry(msg.channel_id)
+									.or_insert_with(Vec::new)
+									.push(RAAMonitorUpdateBlockingAction::from_prev_hop_data(
+										prev_hop,
+									));
+							}
 						}
 
 						// Note that we do not need to push an `actions_blocking_raa_monitor_updates`
@@ -13730,29 +13789,22 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 										.channel_by_id
 										.contains_key(&channel_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. In the future we can
-								// store attribution data in the `ChannelMonitor` and provide it here.
-								self.claim_funds_internal(
-									htlc_update.source,
-									preimage,
-									htlc_update.htlc_value_msat,
-									None,
-									from_onchain,
-									counterparty_node_id,
-									funding_outpoint,
-									channel_id,
-									htlc_update.user_channel_id,
-									None,
-									None,
-									Some(event_id),
-								);
-							}
-							if !we_are_sender {
-								self.chain_monitor.ack_monitor_event(monitor_event_source);
-							}
+							// Claim the funds from the previous hop, if there is one. In the future we can
+							// store attribution data in the `ChannelMonitor` and provide it here.
+							self.claim_funds_internal(
+								htlc_update.source,
+								preimage,
+								htlc_update.htlc_value_msat,
+								None,
+								from_onchain,
+								counterparty_node_id,
+								funding_outpoint,
+								channel_id,
+								htlc_update.user_channel_id,
+								None,
+								None,
+								Some(event_id),
+							);
 						} else {
 							log_trace!(logger, "Failing HTLC from our monitor");
 							let failure_reason = LocalHTLCFailureReason::OnChainTimeout;
@@ -20679,6 +20731,12 @@ impl<
 			downstream_user_channel_id,
 		) in pending_claims_to_replay
 		{
+			// If persistent_monitor_events is enabled, we don't need to explicitly reclaim HTLCs on
+			// startup because we can just wait for the relevant MonitorEvents to be re-provided to us
+			// during runtime.
+			if channel_manager.persistent_monitor_events {
+				continue;
+			}
 			// We use `downstream_closed` in place of `from_onchain` here just as a guess - we
 			// don't remember in the `ChannelMonitor` where we got a preimage from, but if the
 			// channel is closed we just assume that it probably came from an on-chain claim.

lightning/src/ln/channelmanager.rs

@@ -3166,7 +3166,7 @@ pub fn expect_payment_forwarded<CM: AChannelManager, H: NodeHolder<CM = CM>>(
 macro_rules! expect_payment_forwarded {
 	($node: expr, $prev_node: expr, $next_node: expr, $expected_fee: expr, $upstream_force_closed: expr, $downstream_force_closed: expr) => {
 		let mut events = $node.node.get_and_clear_pending_events();
-		assert_eq!(events.len(), 1);
+		assert_eq!(events.len(), 1, "{events:?}");
 		$crate::ln::functional_test_utils::expect_payment_forwarded(
 			events.pop().unwrap(),
 			&$node,
@@ -5673,7 +5673,13 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
 				node_a.node.handle_revoke_and_ack(node_b_id, &bs_revoke_and_ack);
 				check_added_monitors(
 					&node_a,
-					if pending_responding_commitment_signed_dup_monitor.1 { 0 } else { 1 },
+					if pending_responding_commitment_signed_dup_monitor.1
+						&& !node_a.node.test_persistent_monitor_events_enabled()
+					{
+						0
+					} else {
+						1
+					},
 				);
 				if !allow_post_commitment_dance_msgs.1 {
 					assert!(node_a.node.get_and_clear_pending_msg_events().is_empty());

lightning/src/ln/functional_test_utils.rs

@@ -7916,7 +7916,15 @@ fn do_test_onchain_htlc_settlement_after_close(
 		_ => panic!("Unexpected event"),
 	};
 	nodes[1].node.handle_revoke_and_ack(node_c_id, &carol_revocation);
-	check_added_monitors(&nodes[1], 1);
+	if nodes[1].node.test_persistent_monitor_events_enabled() {
+		if broadcast_alice && !go_onchain_before_fulfill {
+			check_added_monitors(&nodes[1], 1);
+		} else {
+			check_added_monitors(&nodes[1], 2);
+		}
+	} else {
+		check_added_monitors(&nodes[1], 1);
+	}
 
 	// If this test requires the force-closed channel to not be on-chain until after the fulfill,
 	// here's where we put said channel's commitment tx on-chain.
@@ -7950,6 +7958,13 @@ fn do_test_onchain_htlc_settlement_after_close(
 			check_spends!(bob_txn[0], chan_ab.3);
 		}
 	}
+	if nodes[1].node.test_persistent_monitor_events_enabled() {
+		if !broadcast_alice || go_onchain_before_fulfill {
+			// In some cases we'll replay the claim via a MonitorEvent and be unable to detect that it's
+			// a duplicate since the inbound edge is on-chain.
+			expect_payment_forwarded!(nodes[1], nodes[0], nodes[2], fee, went_onchain, false);
+		}
+	}
 
 	// Step (6):
 	// Finally, check that Bob broadcasted a preimage-claiming transaction for the HTLC output on the

lightning/src/ln/functional_tests.rs

@@ -927,8 +927,33 @@ fn do_retry_with_no_persist(confirm_before_reload: bool) {
 	let fulfill_msg = htlc_fulfill.update_fulfill_htlcs.remove(0);
 	nodes[1].node.handle_update_fulfill_htlc(node_c_id, fulfill_msg);
 	check_added_monitors(&nodes[1], 1);
-	do_commitment_signed_dance(&nodes[1], &nodes[2], &htlc_fulfill.commitment_signed, false, false);
-	expect_payment_forwarded!(nodes[1], nodes[0], nodes[2], None, true, false);
+	{
+		// Drive the commitment signed dance manually so we can account for the extra monitor
+		// update when persistent monitor events are enabled.
+		let persistent = nodes[1].node.test_persistent_monitor_events_enabled();
+		nodes[1].node.handle_commitment_signed_batch_test(node_c_id, &htlc_fulfill.commitment_signed);
+		check_added_monitors(&nodes[1], 1);
+		let (extra_msg, cs_raa, htlcs) =
+			do_main_commitment_signed_dance(&nodes[1], &nodes[2], false);
+		assert!(htlcs.is_empty());
+		assert!(extra_msg.is_none());
+		// nodes[1] handles nodes[2]'s RAA. When persistent monitor events are enabled, this
+		// triggers the re-provided outbound monitor event, generating an extra preimage update
+		// on the (closed) inbound channel.
+		nodes[1].node.handle_revoke_and_ack(node_c_id, &cs_raa);
+		check_added_monitors(&nodes[1], if persistent { 2 } else { 1 });
+	}
+	if nodes[1].node.test_persistent_monitor_events_enabled() {
+		// The re-provided monitor event generates a duplicate PaymentForwarded against the
+		// closed inbound channel.
+		let events = nodes[1].node.get_and_clear_pending_events();
+		assert_eq!(events.len(), 2);
+		for event in events {
+			assert!(matches!(event, Event::PaymentForwarded { .. }));
+		}
+	} else {
+		expect_payment_forwarded!(nodes[1], nodes[0], nodes[2], None, true, false);
+	}
 
 	if confirm_before_reload {
 		let best_block = nodes[0].blocks.lock().unwrap().last().unwrap().clone();

lightning/src/ln/payment_tests.rs

@@ -1972,6 +1972,13 @@ fn test_reload_node_with_preimage_in_monitor_claims_htlc() {
 		Some(true)
 	);
 
+	if nodes[1].node.test_persistent_monitor_events_enabled() {
+		// Polling messages causes us to re-release the unacked HTLC claim monitor event, which
+		// regenerates a preimage monitor update and forward event below.
+		let msgs = nodes[1].node.get_and_clear_pending_msg_events();
+		assert!(msgs.is_empty());
+	}
+
 	// When the claim is reconstructed during reload, a PaymentForwarded event is generated.
 	// Fetching events triggers the pending monitor update (adding preimage) to be applied.
 	expect_payment_forwarded!(nodes[1], nodes[0], nodes[2], Some(1000), false, false);

lightning/src/ln/reload_tests.rs

@@ -523,6 +523,9 @@ pub struct TestChainMonitor<'a> {
 	pub pause_flush: AtomicBool,
 	/// Buffer of the last 20 monitor updates, most recent first.
 	pub recent_monitor_updates: Mutex<Vec<(ChannelId, ChannelMonitorUpdate)>>,
+	/// When set to `true`, `release_pending_monitor_events` sorts events by `ChannelId` to
+	/// ensure deterministic processing order regardless of HashMap iteration order.
+	pub deterministic_mon_events_order: AtomicBool,
 }
 impl<'a> TestChainMonitor<'a> {
 	pub fn new(
@@ -584,6 +587,7 @@ impl<'a> TestChainMonitor<'a> {
 			write_blocker: Mutex::new(None),
 			pause_flush: AtomicBool::new(false),
 			recent_monitor_updates: Mutex::new(Vec::new()),
+			deterministic_mon_events_order: AtomicBool::new(false),
 		}
 	}
 
@@ -745,7 +749,11 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 			let count = self.chain_monitor.pending_operation_count();
 			self.chain_monitor.flush(count, &self.logger);
 		}
-		return self.chain_monitor.release_pending_monitor_events();
+		let mut events = self.chain_monitor.release_pending_monitor_events();
+		if self.deterministic_mon_events_order.load(Ordering::Acquire) {
+			events.sort_by_key(|(_, channel_id, _, _)| *channel_id);
+		}
+		events
 	}
 
 	fn ack_monitor_event(&self, source: MonitorEventSource) {