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.

We also fix the reload_node test util to use the node's pre-reload config after
restart. The previous behavior was a bit surprising and led to one of this
commit's tests failing.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -18271,6 +18271,76 @@ where
 			}
 		}
 
+		// Remove HTLCs from `forward_htlcs` if they are also present in `decode_update_add_htlcs`.
+		//
+		// In the future, the full set of pending HTLCs will be pulled from `Channel{Monitor}` data and
+		// placed in `ChannelManager::decode_update_add_htlcs` on read, to be handled on the next call
+		// to `process_pending_htlc_forwards`. This is part of a larger effort to remove the requirement
+		// of regularly persisting the `ChannelManager`. The new pipeline is supported for HTLC forwards
+		// received on LDK 0.3+ but not <= 0.2, so prune non-legacy HTLCs from `forward_htlcs`.
+		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
+		});
+		// Add legacy update_adds that were received on LDK <= 0.2 that are not present in the
+		// `decode_update_add_htlcs` map that was rebuilt from `Channel{Monitor}` data, see above
+		// comment.
+		for (scid, legacy_update_adds) in decode_update_add_htlcs_legacy.drain() {
+			match decode_update_add_htlcs.entry(scid) {
+				hash_map::Entry::Occupied(mut update_adds) => {
+					for legacy_update_add in legacy_update_adds {
+						if !update_adds.get().contains(&legacy_update_add) {
+							update_adds.get_mut().push(legacy_update_add);
+						}
+					}
+				},
+				hash_map::Entry::Vacant(entry) => {
+					entry.insert(legacy_update_adds);
+				},
+			}
+		}
+
 		let best_block = BestBlock::new(best_block_hash, best_block_height);
 		let flow = OffersMessageFlow::new(
 			chain_hash,
@@ -18300,7 +18370,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

@@ -508,7 +508,7 @@ 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::ln::outbound_payment::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);