Immediately unblock channels on duplicate claims

When `MonitorUpdateCompletionAction`s were added, we didn't
consider the case of a duplicate claim during normal HTLC
processing (as the handling only had an `if let` rather than a
`match`, which made the branch easy to miss). This can lead to a
channel freezing indefinitely if an HTLC is claimed (without a
`commitment_signed`), the peer disconnects, and then the HTLC is
claimed again, leading to a never-completing
`MonitorUpdateCompletionAction`.

The fix is simple - if we get back an
`UpdateFulfillCommitFetch::DuplicateClaim` when claiming from the
inbound edge, immediately unlock the outbound edge channel with a
new `MonitorUpdateCompletionAction::FreeOtherChannelImmediately`.

Here we implement this fix by actually generating the new variant
when a claim is duplicative.

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -3431,3 +3431,46 @@ fn test_reload_mon_update_completion_actions() {
 	do_test_reload_mon_update_completion_actions(true);
 	do_test_reload_mon_update_completion_actions(false);
 }
+
+#[test]
+fn test_glacial_peer_cant_hang() {
+	// Test that if a peer manages to send an `update_fulfill_htlc` message without a
+	// `commitment_signed`, disconnects, then replays the `update_fulfill_htlc` message it doesn't
+	// result in a channel hang. This was previously broken as the `DuplicateClaim` case wasn't
+	// handled when claiming an HTLC and handling wasn't added when completion actions were added
+	// (which must always complete at some point).
+	let chanmon_cfgs = create_chanmon_cfgs(3);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+	let chan_id_ab = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+	let chan_id_bc = create_announced_chan_between_nodes(&nodes, 1, 2).2;
+
+	// Route a payment from A, through B, to C, then claim it on C. Replay the
+	// `update_fulfill_htlc` twice on B to check that B doesn't hang.
+	let (payment_preimage, payment_hash, ..) = route_payment(&nodes[0], &[&nodes[1], &nodes[2]], 1_000_000);
+
+	nodes[2].node.claim_funds(payment_preimage);
+	check_added_monitors(&nodes[2], 1);
+	expect_payment_claimed!(nodes[2], payment_hash, 1_000_000);
+
+	let cs_updates = get_htlc_update_msgs(&nodes[2], &nodes[1].node.get_our_node_id());
+	nodes[1].node.handle_update_fulfill_htlc(&nodes[2].node.get_our_node_id(), &cs_updates.update_fulfill_htlcs[0]);
+
+	check_added_monitors(&nodes[1], 1);
+	let bs_updates = get_htlc_update_msgs(&nodes[1], &nodes[0].node.get_our_node_id());
+	nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_updates.update_fulfill_htlcs[0]);
+	commitment_signed_dance!(nodes[0], nodes[1], bs_updates.commitment_signed, false);
+	expect_payment_sent!(nodes[0], payment_preimage);
+
+	nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id());
+	nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id());
+
+	let mut reconnect = ReconnectArgs::new(&nodes[1], &nodes[2]);
+	reconnect.pending_htlc_claims = (1, 0);
+	reconnect_nodes(reconnect);
+
+	expect_payment_forwarded!(nodes[1], nodes[0], nodes[2], Some(1000), false, false);
+}

lightning/src/ln/channelmanager.rs

@@ -563,6 +563,7 @@ struct ClaimablePayments {
 /// usually because we're running pre-full-init. They are handled immediately once we detect we are
 /// running normally, and specifically must be processed before any other non-background
 /// [`ChannelMonitorUpdate`]s are applied.
+#[derive(Debug)]
 enum BackgroundEvent {
 	/// Handle a ChannelMonitorUpdate which closes the channel or for an already-closed channel.
 	/// This is only separated from [`Self::MonitorUpdateRegeneratedOnStartup`] as the
@@ -5375,8 +5376,11 @@ where
 			for htlc in sources.drain(..) {
 				if let Err((pk, err)) = self.claim_funds_from_hop(
 					htlc.prev_hop, payment_preimage,
-					|_| Some(MonitorUpdateCompletionAction::PaymentClaimed { payment_hash }))
-				{
+					|_, definitely_duplicate| {
+						debug_assert!(!definitely_duplicate, "We shouldn't claim duplicatively from a payment");
+						Some(MonitorUpdateCompletionAction::PaymentClaimed { payment_hash })
+					}
+				) {
 					if let msgs::ErrorAction::IgnoreError = err.err.action {
 						// We got a temporary failure updating monitor, but will claim the
 						// HTLC when the monitor updating is restored (or on chain).
@@ -5404,7 +5408,7 @@ where
 		}
 	}
 
-	fn claim_funds_from_hop<ComplFunc: FnOnce(Option<u64>) -> Option<MonitorUpdateCompletionAction>>(&self,
+	fn claim_funds_from_hop<ComplFunc: FnOnce(Option<u64>, bool) -> Option<MonitorUpdateCompletionAction>>(&self,
 		prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage, completion_action: ComplFunc)
 	-> Result<(), (PublicKey, MsgHandleErrInternal)> {
 		//TODO: Delay the claimed_funds relaying just like we do outbound relay!
@@ -5414,6 +5418,11 @@ where
 		// `BackgroundEvent`s.
 		let during_init = !self.background_events_processed_since_startup.load(Ordering::Acquire);
 
+		// As we may call handle_monitor_update_completion_actions in rather rare cases, check that
+		// the required mutexes are not held before we start.
+		debug_assert_ne!(self.pending_events.held_by_thread(), LockHeldState::HeldByThread);
+		debug_assert_ne!(self.claimable_payments.held_by_thread(), LockHeldState::HeldByThread);
+
 		{
 			let per_peer_state = self.per_peer_state.read().unwrap();
 			let chan_id = prev_hop.outpoint.to_channel_id();
@@ -5435,25 +5444,58 @@ where
 						let counterparty_node_id = chan.context.get_counterparty_node_id();
 						let fulfill_res = chan.get_update_fulfill_htlc_and_commit(prev_hop.htlc_id, payment_preimage, &self.logger);
 
-						if let UpdateFulfillCommitFetch::NewClaim { htlc_value_msat, monitor_update } = fulfill_res {
-							if let Some(action) = completion_action(Some(htlc_value_msat)) {
-								log_trace!(self.logger, "Tracking monitor update completion action for channel {}: {:?}",
-									chan_id, action);
-								peer_state.monitor_update_blocked_actions.entry(chan_id).or_insert(Vec::new()).push(action);
+						match fulfill_res {
+							UpdateFulfillCommitFetch::NewClaim { htlc_value_msat, monitor_update } => {
+								if let Some(action) = completion_action(Some(htlc_value_msat), false) {
+									log_trace!(self.logger, "Tracking monitor update completion action for channel {}: {:?}",
+										chan_id, action);
+									peer_state.monitor_update_blocked_actions.entry(chan_id).or_insert(Vec::new()).push(action);
+								}
+								if !during_init {
+									handle_new_monitor_update!(self, prev_hop.outpoint, monitor_update, peer_state_lock,
+										peer_state, per_peer_state, chan);
+								} else {
+									// If we're running during init we cannot update a monitor directly -
+									// they probably haven't actually been loaded yet. Instead, push the
+									// monitor update as a background event.
+									self.pending_background_events.lock().unwrap().push(
+										BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+											counterparty_node_id,
+											funding_txo: prev_hop.outpoint,
+											update: monitor_update.clone(),
+										});
+								}
 							}
-							if !during_init {
-								handle_new_monitor_update!(self, prev_hop.outpoint, monitor_update, peer_state_lock,
-									peer_state, per_peer_state, chan);
-							} else {
-								// If we're running during init we cannot update a monitor directly -
-								// they probably haven't actually been loaded yet. Instead, push the
-								// monitor update as a background event.
-								self.pending_background_events.lock().unwrap().push(
-									BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
-										counterparty_node_id,
-										funding_txo: prev_hop.outpoint,
-										update: monitor_update.clone(),
-									});
+							UpdateFulfillCommitFetch::DuplicateClaim {} => {
+								if let Some(action) = completion_action(None, true) {
+									log_trace!(self.logger, "Completing monitor update completion action for channel {} as claim was redundant: {:?}",
+										chan_id, action);
+									mem::drop(peer_state_lock);
+									if let MonitorUpdateCompletionAction::FreeOtherChannelImmediately {
+										downstream_counterparty_and_funding_outpoint
+									} = action {
+										let (node_id, funding_outpoint, blocker) =
+											downstream_counterparty_and_funding_outpoint;
+										if let Some(peer_state_mtx) = per_peer_state.get(&node_id) {
+											let mut peer_state = peer_state_mtx.lock().unwrap();
+											if let Some(blockers) = peer_state.actions_blocking_raa_monitor_updates
+												.get_mut(&funding_outpoint.to_channel_id())
+											{
+												let mut found_blocker = false;
+												blockers.retain(|iter| {
+													if *iter == blocker { found_blocker = true; }
+													*iter != blocker
+												});
+												debug_assert!(found_blocker);
+											}
+										} else {
+											debug_assert!(false);
+										}
+									} else {
+										debug_assert!(false,
+											"Duplicate claims should always free another channel immediately");
+									}
+								}
 							}
 						}
 					}
@@ -5501,7 +5543,7 @@ where
 		// `ChannelMonitor` we've provided the above update to. Instead, note that `Event`s are
 		// generally always allowed to be duplicative (and it's specifically noted in
 		// `PaymentForwarded`).
-		self.handle_monitor_update_completion_actions(completion_action(None));
+		self.handle_monitor_update_completion_actions(completion_action(None, false));
 		Ok(())
 	}
 
@@ -5531,33 +5573,72 @@ where
 			HTLCSource::PreviousHopData(hop_data) => {
 				let prev_outpoint = hop_data.outpoint;
 				let completed_blocker = RAAMonitorUpdateBlockingAction::from_prev_hop_data(&hop_data);
+				#[cfg(debug_assertions)]
+				let claiming_chan_funding_outpoint = hop_data.outpoint;
 				let res = self.claim_funds_from_hop(hop_data, payment_preimage,
-					|htlc_claim_value_msat| {
+					|htlc_claim_value_msat, definitely_duplicate| {
 						if let Some(forwarded_htlc_value) = forwarded_htlc_value_msat {
 							let fee_earned_msat = if let Some(claimed_htlc_value) = htlc_claim_value_msat {
 								Some(claimed_htlc_value - forwarded_htlc_value)
 							} else { None };
 
-							Some(MonitorUpdateCompletionAction::EmitEventAndFreeOtherChannel {
-								event: events::Event::PaymentForwarded {
-									fee_earned_msat,
-									claim_from_onchain_tx: from_onchain,
-									prev_channel_id: Some(prev_outpoint.to_channel_id()),
-									next_channel_id: Some(next_channel_outpoint.to_channel_id()),
-									outbound_amount_forwarded_msat: forwarded_htlc_value_msat,
-								},
-								downstream_counterparty_and_funding_outpoint:
-									if let Some(node_id) = next_channel_counterparty_node_id {
-										Some((node_id, next_channel_outpoint, completed_blocker))
-									} else {
-										// We can only get `None` here if we are processing a
-										// `ChannelMonitor`-originated event, in which case we
-										// don't care about ensuring we wake the downstream
-										// channel's monitor updating - the channel is already
-										// closed.
-										None
+							let chan_to_release =
+								if let Some(node_id) = next_channel_counterparty_node_id {
+									Some((node_id, next_channel_outpoint, completed_blocker))
+								} else {
+									// We can only get `None` here if we are processing a
+									// `ChannelMonitor`-originated event, in which case we
+									// don't care about ensuring we wake the downstream
+									// channel's monitor updating - the channel is already
+									// closed.
+									None
+								};
+
+							if definitely_duplicate && startup_replay {
+								// On startup we may get 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
+								// in the background.
+								#[cfg(debug_assertions)] {
+									let background_events = self.pending_background_events.lock().unwrap();
+									// There should be a `BackgroundEvent` pending...
+									assert!(background_events.iter().any(|ev| {
+										match ev {
+											// to apply a monitor update that blocked channel,
+											BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+												funding_txo, ..
+											} => *funding_txo == claiming_chan_funding_outpoint,
+											// or the channel we'd unblock is already closed,
+											BackgroundEvent::ClosedMonitorUpdateRegeneratedOnStartup((funding_txo, ..))
+												=> *funding_txo == next_channel_outpoint,
+											// or the monitor update has completed and will unblock
+											// immediately once we get going.
+											BackgroundEvent::MonitorUpdatesComplete {
+												channel_id, ..
+											} =>
+												*channel_id == claiming_chan_funding_outpoint.to_channel_id(),
+										}
+									}), "{:?}", *background_events);
+								}
+								None
+							} else if definitely_duplicate {
+								if let Some(other_chan) = chan_to_release {
+									Some(MonitorUpdateCompletionAction::FreeOtherChannelImmediately {
+										downstream_counterparty_and_funding_outpoint: other_chan,
+									})
+								} else { None }
+							} else {
+								Some(MonitorUpdateCompletionAction::EmitEventAndFreeOtherChannel {
+									event: events::Event::PaymentForwarded {
+										fee_earned_msat,
+										claim_from_onchain_tx: from_onchain,
+										prev_channel_id: Some(prev_outpoint.to_channel_id()),
+										next_channel_id: Some(next_channel_outpoint.to_channel_id()),
+										outbound_amount_forwarded_msat: forwarded_htlc_value_msat,
 									},
-							})
+									downstream_counterparty_and_funding_outpoint: chan_to_release,
+								})
+							}
 						} else { None }
 					});
 				if let Err((pk, err)) = res {
@@ -5574,6 +5655,10 @@ where
 	}
 
 	fn handle_monitor_update_completion_actions<I: IntoIterator<Item=MonitorUpdateCompletionAction>>(&self, actions: I) {
+		debug_assert_ne!(self.pending_events.held_by_thread(), LockHeldState::HeldByThread);
+		debug_assert_ne!(self.claimable_payments.held_by_thread(), LockHeldState::HeldByThread);
+		debug_assert_ne!(self.per_peer_state.held_by_thread(), LockHeldState::HeldByThread);
+
 		for action in actions.into_iter() {
 			match action {
 				MonitorUpdateCompletionAction::PaymentClaimed { payment_hash } => {