Persistent monitor events for HTLC forward claims

XXX

Author: valentinewallace

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -3619,14 +3619,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,
+		}
 	}
 }
 
@@ -9635,16 +9644,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
@@ -9977,6 +10026,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 {
@@ -12691,23 +12741,28 @@ 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 !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`
@@ -13709,29 +13764,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. 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_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. 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_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;
@@ -20658,6 +20706,9 @@ impl<
 			downstream_user_channel_id,
 		) in pending_claims_to_replay
 		{
+			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) {