Rebuild manager forwarded htlcs maps from Channels

We have an overarching goal of (mostly) getting rid of ChannelManager
persistence and rebuilding the ChannelManager's state from existing
ChannelMonitors, due to issues when the two structs are out-of-sync on restart.
The main issue that can arise is channel force closure.

Here we start this process by rebuilding
ChannelManager::decode_update_add_htlcs, forward_htlcs, and
pending_intercepted_htlcs from Channel data, which will soon be included in the
ChannelMonitors as part of a different series of PRs.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -11551,6 +11551,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 
 			if !new_intercept_events.is_empty() {
 				let mut events = self.pending_events.lock().unwrap();
+				new_intercept_events.retain(|new_ev| !events.contains(new_ev));
 				events.append(&mut new_intercept_events);
 			}
 		}
@@ -18265,6 +18266,58 @@ where
 				);
 			}
 		}
+		// Remove forwarded HTLCs if they are also present in `decode_update_add_htlcs`. HTLCs present
+		// in both maps will be re-decoded from `decode_update_add_htlcs` and placed in
+		// `ChannelManager::forward_htlcs` on the next call to `process_pending_htlc_forwards`.
+		//
+		// In future versions we'll stop (de)serializing `forward_htlcs` as part of getting rid of the
+		// requirement to regularly persist the `ChannelManager`, so here we move in that direction by
+		// preferring to keep HTLCs in the new map that is rebuilt from `Channel{Monitor}` data.
+		forward_htlcs_legacy.retain(|scid, pending_fwds| {
+			for fwd in pending_fwds {
+				let (prev_scid, prev_htlc_id) = match fwd {
+					HTLCForwardInfo::AddHTLC(htlc) => {
+						(htlc.prev_outbound_scid_alias, htlc.prev_htlc_id)
+					},
+					HTLCForwardInfo::FailHTLC { htlc_id, .. }
+					| HTLCForwardInfo::FailMalformedHTLC { htlc_id, .. } => (*scid, *htlc_id),
+				};
+				if let Some(pending_update_adds) = decode_update_add_htlcs.get_mut(&prev_scid) {
+					if pending_update_adds
+						.iter()
+						.any(|update_add| update_add.htlc_id == prev_htlc_id)
+					{
+						return false;
+					}
+				}
+			}
+			true
+		});
+		// Remove intercepted HTLC forwards if they are also present in `decode_update_add_htlcs`. See
+		// the above comment.
+		pending_intercepted_htlcs_legacy.retain(|id, fwd| {
+			let prev_scid = fwd.prev_outbound_scid_alias;
+			if let Some(pending_update_adds) = decode_update_add_htlcs.get_mut(&prev_scid) {
+				if pending_update_adds
+					.iter()
+					.any(|update_add| update_add.htlc_id == fwd.prev_htlc_id)
+				{
+					pending_events_read.retain(
+						|(ev, _)| !matches!(ev, Event::HTLCIntercepted { intercept_id, .. } if intercept_id == id),
+					);
+					return false;
+				}
+			}
+			if !pending_events_read.iter().any(
+				|(ev, _)| matches!(ev, Event::HTLCIntercepted { intercept_id, .. } if intercept_id == id),
+			) {
+				match create_htlc_intercepted_event(*id, &fwd) {
+					Ok(ev) => pending_events_read.push_back((ev, None)),
+					Err(()) => debug_assert!(false),
+				}
+			}
+			true
+		});
 
 		let best_block = BestBlock::new(best_block_hash, best_block_height);
 		let flow = OffersMessageFlow::new(
@@ -18295,7 +18348,7 @@ where
 			pending_intercepted_htlcs: Mutex::new(pending_intercepted_htlcs_legacy),
 
 			forward_htlcs: Mutex::new(forward_htlcs_legacy),
-			decode_update_add_htlcs: Mutex::new(decode_update_add_htlcs_legacy),
+			decode_update_add_htlcs: Mutex::new(decode_update_add_htlcs),
 			claimable_payments: Mutex::new(ClaimablePayments {
 				claimable_payments,
 				pending_claiming_payments: pending_claiming_payments.unwrap(),

lightning/src/ln/functional_test_utils.rs

@@ -1382,9 +1382,10 @@ macro_rules! reload_node {
 		$node.onion_messenger.set_async_payments_handler(&$new_channelmanager);
 	};
 	($node: expr, $chanman_encoded: expr, $monitors_encoded: expr, $persister: ident, $new_chain_monitor: ident, $new_channelmanager: ident) => {
+		let config = $node.node.get_current_config();
 		reload_node!(
 			$node,
-			test_default_channel_config(),
+			config,
 			$chanman_encoded,
 			$monitors_encoded,
 			$persister,

lightning/src/ln/reload_tests.rs

@@ -20,6 +20,7 @@ use crate::chain::transaction::OutPoint;
 use crate::events::{ClosureReason, Event, HTLCHandlingFailureType};
 use crate::ln::channelmanager::{ChannelManager, ChannelManagerReadArgs, PaymentId, RecipientOnionFields, RAACommitmentOrder};
 use crate::ln::msgs;
+use crate::ln::outbound_payment::Retry;
 use crate::ln::types::ChannelId;
 use crate::ln::msgs::{BaseMessageHandler, ChannelMessageHandler, RoutingMessageHandler, ErrorAction, MessageSendEvent};
 use crate::util::test_channel_signer::TestChannelSigner;
@@ -508,7 +509,6 @@ fn test_manager_serialize_deserialize_inconsistent_monitor() {
 
 #[cfg(feature = "std")]
 fn do_test_data_loss_protect(reconnect_panicing: bool, substantially_old: bool, not_stale: bool) {
-	use crate::ln::channelmanager::Retry;
 	use crate::types::string::UntrustedString;
 	// When we get a data_loss_protect proving we're behind, we immediately panic as the
 	// chain::Watch API requirements have been violated (e.g. the user restored from a backup). The
@@ -1173,6 +1173,91 @@ fn removed_payment_no_manager_persistence() {
 	expect_payment_failed!(nodes[0], payment_hash, false);
 }
 
+#[test]
+fn manager_persisted_pre_outbound_edge_forward() {
+	do_manager_persisted_pre_outbound_edge_forward(false);
+}
+
+#[test]
+fn manager_persisted_pre_outbound_edge_intercept_forward() {
+	do_manager_persisted_pre_outbound_edge_forward(true);
+}
+
+fn do_manager_persisted_pre_outbound_edge_forward(intercept_htlc: bool) {
+	let chanmon_cfgs = create_chanmon_cfgs(3);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+	let persister;
+	let new_chain_monitor;
+	let mut intercept_forwards_config = test_default_channel_config();
+	intercept_forwards_config.accept_intercept_htlcs = true;
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, Some(intercept_forwards_config), None]);
+	let nodes_1_deserialized;
+	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+	let chan_id_1 = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+	let chan_id_2 = create_announced_chan_between_nodes(&nodes, 1, 2).2;
+
+	// Lock in the HTLC from node_a <> node_b.
+	let amt_msat = 5000;
+	let (mut route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[2], amt_msat);
+	if intercept_htlc {
+		route.paths[0].hops[1].short_channel_id = nodes[1].node.get_intercept_scid();
+	}
+	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 updates = get_htlc_update_msgs(&nodes[0], &nodes[1].node.get_our_node_id());
+	nodes[1].node.handle_update_add_htlc(nodes[0].node.get_our_node_id(), &updates.update_add_htlcs[0]);
+	do_commitment_signed_dance(&nodes[1], &nodes[0], &updates.commitment_signed, false, false);
+
+	// Decode the HTLC onion but don't forward it to the next hop, such that the HTLC ends up in
+	// `ChannelManager::forward_htlcs` or `ChannelManager::pending_intercepted_htlcs`.
+	nodes[1].node.test_process_pending_update_add_htlcs();
+
+	// Disconnect peers and reload the forwarding node_b.
+	nodes[0].node.peer_disconnected(nodes[1].node.get_our_node_id());
+	nodes[2].node.peer_disconnected(nodes[1].node.get_our_node_id());
+
+	let node_b_encoded = nodes[1].node.encode();
+
+	let chan_0_monitor_serialized = get_monitor!(nodes[1], chan_id_1).encode();
+	let chan_1_monitor_serialized = get_monitor!(nodes[1], chan_id_2).encode();
+	reload_node!(nodes[1], node_b_encoded, &[&chan_0_monitor_serialized, &chan_1_monitor_serialized], persister, new_chain_monitor, nodes_1_deserialized);
+
+	reconnect_nodes(ReconnectArgs::new(&nodes[1], &nodes[0]));
+	let mut args_b_c = ReconnectArgs::new(&nodes[1], &nodes[2]);
+	args_b_c.send_channel_ready = (true, true);
+	args_b_c.send_announcement_sigs = (true, true);
+	reconnect_nodes(args_b_c);
+
+	// Forward the HTLC and ensure we can claim it post-reload.
+	nodes[1].node.process_pending_htlc_forwards();
+
+	if intercept_htlc {
+		let events = nodes[1].node.get_and_clear_pending_events();
+		assert_eq!(events.len(), 1);
+		let (intercept_id, expected_outbound_amt_msat) = match events[0] {
+			Event::HTLCIntercepted { intercept_id, expected_outbound_amount_msat, .. } => {
+				(intercept_id, expected_outbound_amount_msat)
+			},
+			_ => panic!()
+		};
+		nodes[1].node.forward_intercepted_htlc(intercept_id, &chan_id_2,
+			nodes[2].node.get_our_node_id(), expected_outbound_amt_msat).unwrap();
+		nodes[1].node.process_pending_htlc_forwards();
+	}
+	check_added_monitors(&nodes[1], 1);
+
+	let updates = get_htlc_update_msgs(&nodes[1], &nodes[2].node.get_our_node_id());
+	nodes[2].node.handle_update_add_htlc(nodes[1].node.get_our_node_id(), &updates.update_add_htlcs[0]);
+	do_commitment_signed_dance(&nodes[2], &nodes[1], &updates.commitment_signed, false, false);
+	expect_and_process_pending_htlcs(&nodes[2], false);
+
+	expect_payment_claimable!(nodes[2], payment_hash, payment_secret, amt_msat, None, nodes[2].node.get_our_node_id());
+	let path: &[&[_]] = &[&[&nodes[1], &nodes[2]]];
+	do_claim_payment_along_route(ClaimAlongRouteArgs::new(&nodes[0], path, payment_preimage));
+	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
+}
+
 #[test]
 fn test_reload_partial_funding_batch() {
 	let chanmon_cfgs = create_chanmon_cfgs(3);