Implement non-strict forwarding

This change implements non-strict forwarding, allowing the node to
forward an HTLC along a channel other than the one specified
by short_channel_id in the onion message, so long as the receiver has
the same node public key intended by short_channel_id
([BOLT](https://github.com/lightning/bolts/blob/57ce4b1e05c996fa649f00dc13521f6d496a288f/04-onion-routing.md#non-strict-forwarding)).
This can improve payment reliability when there are multiple channels
with the same peer e.g. when outbound liquidity is replenished by
opening a new channel.

The implemented forwarding strategy now chooses the channel with the
lowest outbound liquidity that can forward an HTLC to maximize the
probability of being able to successfully forward a subsequent HTLC.

Fixes #1278.

lightning/src/ln/channelmanager.rs

@@ -4895,8 +4895,8 @@ where
 				if short_chan_id != 0 {
 					let mut forwarding_counterparty = None;
 					macro_rules! forwarding_channel_not_found {
-						() => {
-							for forward_info in pending_forwards.drain(..) {
+						($forward_infos:expr) => {
+							for forward_info in $forward_infos {
 								match forward_info {
 									HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
 										prev_short_channel_id, prev_htlc_id, prev_channel_id, prev_funding_outpoint,
@@ -5004,34 +5004,33 @@ where
 					let (counterparty_node_id, forward_chan_id) = match chan_info_opt {
 						Some((cp_id, chan_id)) => (cp_id, chan_id),
 						None => {
-							forwarding_channel_not_found!();
+							forwarding_channel_not_found!(pending_forwards.drain(..));
 							continue;
 						}
 					};
 					forwarding_counterparty = Some(counterparty_node_id);
 					let per_peer_state = self.per_peer_state.read().unwrap();
 					let peer_state_mutex_opt = per_peer_state.get(&counterparty_node_id);
 					if peer_state_mutex_opt.is_none() {
-						forwarding_channel_not_found!();
+						forwarding_channel_not_found!(pending_forwards.drain(..));
 						continue;
 					}
 					let mut peer_state_lock = peer_state_mutex_opt.unwrap().lock().unwrap();
 					let peer_state = &mut *peer_state_lock;
-					if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
-						let logger = WithChannelContext::from(&self.logger, &chan.context, None);
-						for forward_info in pending_forwards.drain(..) {
+					let mut draining_pending_forwards = pending_forwards.drain(..);
+					loop {
+						let maybe_forward_info = draining_pending_forwards.next();
+						if let Some(forward_info) = maybe_forward_info {
 							let queue_fail_htlc_res = match forward_info {
 								HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
 									prev_short_channel_id, prev_htlc_id, prev_channel_id, prev_funding_outpoint,
 									prev_user_channel_id, forward_info: PendingHTLCInfo {
 										incoming_shared_secret, payment_hash, outgoing_amt_msat, outgoing_cltv_value,
 										routing: PendingHTLCRouting::Forward {
-											onion_packet, blinded, ..
+											ref onion_packet, blinded, ..
 										}, skimmed_fee_msat, ..
 									},
 								}) => {
-									let logger = WithChannelContext::from(&self.logger, &chan.context, Some(payment_hash));
-									log_trace!(logger, "Adding HTLC from short id {} with payment_hash {} to channel with short id {} after delay", prev_short_channel_id, &payment_hash, short_chan_id);
 									let htlc_source = HTLCSource::PreviousHopData(HTLCPreviousHopData {
 										short_channel_id: prev_short_channel_id,
 										user_channel_id: Some(prev_user_channel_id),
@@ -5051,44 +5050,86 @@ where
 											&self.secp_ctx, b.inbound_blinding_point, &encrypted_tlvs_ss
 										).ok()
 									});
-									if let Err(e) = chan.queue_add_htlc(outgoing_amt_msat,
-										payment_hash, outgoing_cltv_value, htlc_source.clone(),
-										onion_packet, skimmed_fee_msat, next_blinding_point, &self.fee_estimator,
-										&&logger)
-									{
-										if let ChannelError::Ignore(msg) = e {
-											log_trace!(logger, "Failed to forward HTLC with payment_hash {}: {}", &payment_hash, msg);
+
+									// Forward the HTLC over the most appropriate channel with the corresponding peer,
+									// applying non-strict forwarding.
+									// The channel with the least amount of outbound liquidity will be used to maximize the
+									// probability of being able to successfully forward a subsequent HTLC.
+									let mut channels_with_peer = peer_state.channel_by_id.values_mut().filter_map(|phase| match phase {
+										ChannelPhase::Funded(chan) => Some(chan),
+										_ => None,
+									}).collect::<Vec<&mut Channel<_>>>();
+									channels_with_peer.sort_by_key(|chan| chan.context.get_available_balances(&self.fee_estimator).outbound_capacity_msat);
+									let successfully_added = channels_with_peer.iter_mut().any(|chan| {
+										let logger = WithChannelContext::from(&self.logger, &chan.context, Some(payment_hash));
+										let add_result = chan.queue_add_htlc(outgoing_amt_msat, payment_hash,
+											outgoing_cltv_value, htlc_source.clone(), onion_packet.clone(),
+											skimmed_fee_msat, next_blinding_point, &self.fee_estimator, &&logger);
+										match add_result {
+											Ok(_) => {
+												log_trace!(logger, "Forwarding HTLC from SCID {} with payment_hash {} and specified next hop SCID {} over channel {} with corresponding peer {}",
+													prev_short_channel_id, &payment_hash, short_chan_id, chan.context.channel_id(), &counterparty_node_id);
+											},
+											Err(ChannelError::Ignore(ref msg)) => {
+												log_trace!(logger, "Not forwarding HTLC with payment_hash {} over channel {} with peer {}: {}. Will attempt other channels with the same peer if possible.",
+													&payment_hash, chan.context.channel_id(), &counterparty_node_id, msg);
+											},
+											Err(_) => {
+												panic!("Stated return value requirements in send_htlc() were not met");
+											},
+										}
+										add_result.is_ok()
+									});
+
+									if !successfully_added {
+										log_trace!(self.logger, "Failed to forward HTLC with payment_hash {} to peer {}", &payment_hash, &counterparty_node_id);
+										if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
+											let (failure_code, data) = self.get_htlc_temp_fail_err_and_data(0x1000|7, short_chan_id, chan);
+											failed_forwards.push((htlc_source, payment_hash,
+												HTLCFailReason::reason(failure_code, data),
+												HTLCDestination::NextHopChannel { node_id: Some(chan.context.get_counterparty_node_id()), channel_id: forward_chan_id }
+											));
 										} else {
-											panic!("Stated return value requirements in send_htlc() were not met");
+											forwarding_channel_not_found!(core::iter::once(forward_info).chain(draining_pending_forwards));
+											break;
 										}
-										let (failure_code, data) = self.get_htlc_temp_fail_err_and_data(0x1000|7, short_chan_id, chan);
-										failed_forwards.push((htlc_source, payment_hash,
-											HTLCFailReason::reason(failure_code, data),
-											HTLCDestination::NextHopChannel { node_id: Some(chan.context.get_counterparty_node_id()), channel_id: forward_chan_id }
-										));
-										continue;
 									}
 									None
 								},
 								HTLCForwardInfo::AddHTLC { .. } => {
 									panic!("short_channel_id != 0 should imply any pending_forward entries are of type Forward");
 								},
-								HTLCForwardInfo::FailHTLC { htlc_id, err_packet } => {
-									log_trace!(logger, "Failing HTLC back to channel with short id {} (backward HTLC ID {}) after delay", short_chan_id, htlc_id);
-									Some((chan.queue_fail_htlc(htlc_id, err_packet, &&logger), htlc_id))
+								HTLCForwardInfo::FailHTLC { htlc_id, ref err_packet } => {
+									if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
+										let logger = WithChannelContext::from(&self.logger, &chan.context, None);
+										log_trace!(logger, "Failing HTLC back to channel with short id {} (backward HTLC ID {}) after delay", short_chan_id, htlc_id);
+										Some((chan.queue_fail_htlc(htlc_id, err_packet.clone(), &&logger), htlc_id))
+									} else {
+										forwarding_channel_not_found!(core::iter::once(forward_info).chain(draining_pending_forwards));
+										break;
+									}
 								},
 								HTLCForwardInfo::FailMalformedHTLC { htlc_id, failure_code, sha256_of_onion } => {
-									log_trace!(logger, "Failing malformed HTLC back to channel with short id {} (backward HTLC ID {}) after delay", short_chan_id, htlc_id);
-									let res = chan.queue_fail_malformed_htlc(
-										htlc_id, failure_code, sha256_of_onion, &&logger
-									);
-									Some((res, htlc_id))
+									if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
+										let logger = WithChannelContext::from(&self.logger, &chan.context, None);
+										log_trace!(logger, "Failing malformed HTLC back to channel with short id {} (backward HTLC ID {}) after delay", short_chan_id, htlc_id);
+										let res = chan.queue_fail_malformed_htlc(
+											htlc_id, failure_code, sha256_of_onion, &&logger
+										);
+										Some((res, htlc_id))
+									} else {
+										forwarding_channel_not_found!(core::iter::once(forward_info).chain(draining_pending_forwards));
+										break;
+									}
 								},
 							};
 							if let Some((queue_fail_htlc_res, htlc_id)) = queue_fail_htlc_res {
 								if let Err(e) = queue_fail_htlc_res {
 									if let ChannelError::Ignore(msg) = e {
-										log_trace!(logger, "Failed to fail HTLC with ID {} backwards to short_id {}: {}", htlc_id, short_chan_id, msg);
+										if let Some(ChannelPhase::Funded(ref mut chan)) = peer_state.channel_by_id.get_mut(&forward_chan_id) {
+											let logger = WithChannelContext::from(&self.logger, &chan.context, None);
+											log_trace!(logger, "Failed to fail HTLC with ID {} backwards to short_id {}: {}", htlc_id, short_chan_id, msg);
+										}
 									} else {
 										panic!("Stated return value requirements in queue_fail_{{malformed_}}htlc() were not met");
 									}
@@ -5098,10 +5139,9 @@ where
 									continue;
 								}
 							}
+						} else {
+							break;
 						}
-					} else {
-						forwarding_channel_not_found!();
-						continue;
 					}
 				} else {
 					'next_forwardable_htlc: for forward_info in pending_forwards.drain(..) {

lightning/src/ln/payment_tests.rs

@@ -2053,22 +2053,28 @@ fn accept_underpaying_htlcs_config() {
 fn do_accept_underpaying_htlcs_config(num_mpp_parts: usize) {
 	let chanmon_cfgs = create_chanmon_cfgs(3);
 	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+	let max_in_flight_percent = 10;
 	let mut intercept_forwards_config = test_default_channel_config();
 	intercept_forwards_config.accept_intercept_htlcs = true;
+	intercept_forwards_config.channel_handshake_config.max_inbound_htlc_value_in_flight_percent_of_channel = max_in_flight_percent;
 	let mut underpay_config = test_default_channel_config();
 	underpay_config.channel_config.accept_underpaying_htlcs = true;
+	underpay_config.channel_handshake_config.max_inbound_htlc_value_in_flight_percent_of_channel = max_in_flight_percent;
 	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, Some(intercept_forwards_config), Some(underpay_config)]);
 	let nodes = create_network(3, &node_cfgs, &node_chanmgrs);
 
+	let amt_msat = 900_000;
+
 	let mut chan_ids = Vec::new();
 	for _ in 0..num_mpp_parts {
-		let _ = create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 10_000, 0);
-		let channel_id = create_unannounced_chan_between_nodes_with_value(&nodes, 1, 2, 2_000_000, 0).0.channel_id;
+		// We choose the channel size so that there can be at most one part pending on each channel.
+		let channel_size = amt_msat / 1000 / num_mpp_parts as u64 * 100 / max_in_flight_percent as u64 + 100;
+		let _ = create_announced_chan_between_nodes_with_value(&nodes, 0, 1, channel_size, 0);
+		let channel_id = create_unannounced_chan_between_nodes_with_value(&nodes, 1, 2, channel_size, 0).0.channel_id;
 		chan_ids.push(channel_id);
 	}
 
 	// Send the initial payment.
-	let amt_msat = 900_000;
 	let skimmed_fee_msat = 20;
 	let mut route_hints = Vec::new();
 	for _ in 0..num_mpp_parts {
@@ -4098,7 +4104,7 @@ fn do_test_payment_metadata_consistency(do_reload: bool, do_modify: bool) {
 
 	// Create a new channel between C and D as A will refuse to retry on the existing one because
 	// it just failed.
-	let chan_id_cd_2 = create_announced_chan_between_nodes_with_value(&nodes, 2, 3, 1_000_000, 0).2;
+	create_announced_chan_between_nodes_with_value(&nodes, 2, 3, 1_000_000, 0);
 
 	// Now retry the failed HTLC.
 	nodes[0].node.process_pending_htlc_forwards();
@@ -4110,6 +4116,7 @@ fn do_test_payment_metadata_consistency(do_reload: bool, do_modify: bool) {
 	expect_pending_htlcs_forwardable!(nodes[2]);
 	check_added_monitors(&nodes[2], 1);
 	let cs_forward = SendEvent::from_node(&nodes[2]);
+	let cd_channel_used = cs_forward.msgs[0].channel_id;
 	nodes[3].node.handle_update_add_htlc(&nodes[2].node.get_our_node_id(), &cs_forward.msgs[0]);
 	commitment_signed_dance!(nodes[3], nodes[2], cs_forward.commitment_msg, false, true);
 
@@ -4129,7 +4136,7 @@ fn do_test_payment_metadata_consistency(do_reload: bool, do_modify: bool) {
 		nodes[2].node.handle_update_fail_htlc(&nodes[3].node.get_our_node_id(), &ds_fail.update_fail_htlcs[0]);
 		commitment_signed_dance!(nodes[2], nodes[3], ds_fail.commitment_signed, false, true);
 		expect_pending_htlcs_forwardable_conditions(nodes[2].node.get_and_clear_pending_events(),
-			&[HTLCDestination::NextHopChannel { node_id: Some(nodes[3].node.get_our_node_id()), channel_id: chan_id_cd_2 }]);
+			&[HTLCDestination::NextHopChannel { node_id: Some(nodes[3].node.get_our_node_id()), channel_id: cd_channel_used }]);
 	} else {
 		expect_pending_htlcs_forwardable!(nodes[3]);
 		expect_payment_claimable!(nodes[3], payment_hash, payment_secret, amt_msat);
@@ -4294,3 +4301,94 @@ fn peel_payment_onion_custom_tlvs() {
 		_ => panic!()
 	}
 }
+
+#[test]
+fn test_non_strict_forwarding() {
+	let chanmon_cfgs = create_chanmon_cfgs(3);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+	let mut config = test_default_channel_config();
+	config.channel_handshake_config.max_inbound_htlc_value_in_flight_percent_of_channel = 100;
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[Some(config), Some(config), Some(config)]);
+	let nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+	// Create a routing node with two outbound channels, each of which can forward 2 payments of
+	// the given value.
+	let payment_value = 1_500_000;
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 100_000, 0);
+	let (chan_update_1, _, channel_id_1, _) = create_announced_chan_between_nodes_with_value(&nodes, 1, 2, 4_950, 0);
+	let (chan_update_2, _, channel_id_2, _) = create_announced_chan_between_nodes_with_value(&nodes, 1, 2, 5_000, 0);
+
+	// Create a route once.
+	let payment_params = PaymentParameters::from_node_id(nodes[2].node.get_our_node_id(), TEST_FINAL_CLTV)
+		.with_bolt11_features(nodes[2].node.bolt11_invoice_features()).unwrap();
+	let route_params = RouteParameters::from_payment_params_and_value(payment_params, payment_value);
+	let route = functional_test_utils::get_route(&nodes[0], &route_params).unwrap();
+
+	// Send 4 payments over the same route.
+	for i in 0..4 {
+		let (payment_preimage, payment_hash, payment_secret) = get_payment_preimage_hash(&nodes[2], Some(payment_value), None);
+		nodes[0].node.send_payment_with_route(&route, payment_hash,
+			RecipientOnionFields::secret_only(payment_secret), PaymentId(payment_hash.0)).unwrap();
+		check_added_monitors!(nodes[0], 1);
+		let mut msg_events = nodes[0].node.get_and_clear_pending_msg_events();
+		assert_eq!(msg_events.len(), 1);
+		let mut send_event = SendEvent::from_event(msg_events.remove(0));
+		nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &send_event.msgs[0]);
+		commitment_signed_dance!(nodes[1], nodes[0], &send_event.commitment_msg, false);
+
+		expect_pending_htlcs_forwardable!(nodes[1]);
+		check_added_monitors!(nodes[1], 1);
+		msg_events = nodes[1].node.get_and_clear_pending_msg_events();
+		assert_eq!(msg_events.len(), 1);
+		send_event = SendEvent::from_event(msg_events.remove(0));
+		// The HTLC will be forwarded over the most appropriate channel with the corresponding peer,
+		// applying non-strict forwarding.
+		// The channel with the least amount of outbound liquidity will be used to maximize the
+		// probability of being able to successfully forward a subsequent HTLC.
+		assert_eq!(send_event.msgs[0].channel_id, if i < 2 {
+			channel_id_1
+		} else {
+			channel_id_2
+		});
+		nodes[2].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &send_event.msgs[0]);
+		commitment_signed_dance!(nodes[2], nodes[1], &send_event.commitment_msg, false);
+
+		expect_pending_htlcs_forwardable!(nodes[2]);
+		let events = nodes[2].node.get_and_clear_pending_events();
+		assert_eq!(events.len(), 1);
+		assert!(matches!(events[0], Event::PaymentClaimable { .. }));
+
+		claim_payment_along_route(
+			ClaimAlongRouteArgs::new(&nodes[0], &[&[&nodes[1], &nodes[2]]], payment_preimage)
+		);
+	}
+
+	// Send a 5th payment which will fail.
+	let (_, payment_hash, payment_secret) = get_payment_preimage_hash(&nodes[2], Some(payment_value), None);
+	nodes[0].node.send_payment_with_route(&route, payment_hash,
+		RecipientOnionFields::secret_only(payment_secret), PaymentId(payment_hash.0)).unwrap();
+	check_added_monitors!(nodes[0], 1);
+	let mut msg_events = nodes[0].node.get_and_clear_pending_msg_events();
+	assert_eq!(msg_events.len(), 1);
+	let mut send_event = SendEvent::from_event(msg_events.remove(0));
+	nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &send_event.msgs[0]);
+	commitment_signed_dance!(nodes[1], nodes[0], &send_event.commitment_msg, false);
+
+	expect_pending_htlcs_forwardable!(nodes[1]);
+	check_added_monitors!(nodes[1], 1);
+	let routed_scid = route.paths[0].hops[1].short_channel_id;
+	let routed_channel_id = match routed_scid {
+		scid if scid == chan_update_1.contents.short_channel_id => channel_id_1,
+		scid if scid == chan_update_2.contents.short_channel_id => channel_id_2,
+		_ => panic!("Unexpected short channel id in route"),
+	};
+	// The failure to forward will refer to the channel given in the onion.
+	expect_pending_htlcs_forwardable_conditions(nodes[1].node.get_and_clear_pending_events(),
+		&[HTLCDestination::NextHopChannel { node_id: Some(nodes[2].node.get_our_node_id()), channel_id: routed_channel_id }]);
+
+	let updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
+	nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &updates.update_fail_htlcs[0]);
+	commitment_signed_dance!(nodes[0], nodes[1], updates.commitment_signed, false);
+	let events = nodes[0].node.get_and_clear_pending_events();
+	expect_payment_failed_conditions_event(events, payment_hash, false, PaymentFailedConditions::new().blamed_scid(routed_scid));
+}