Prefer legacy forward maps on manager read

We are working on removing the requirement of regularly persisting the
ChannelManager, and as a result began reconstructing the manager's forwards
maps from Channel data on startup in a recent PR, see
cb398f6 and parent commits.

At the time, we implemented ChannelManager::read to prefer to use the newly
reconstructed maps, partly to ensure we have test coverage of the new maps'
usage. This resulted in a lot of code that would deduplicate HTLCs that were
present in the old maps to avoid redundant HTLC handling/duplicate forwards,
adding extra complexity.

Instead, always use the old maps in prod, but randomly use the newly
reconstructed maps in testing, to exercise the new codepaths (see
reconstruct_manager_from_monitors in ChannelManager::read).

Author: valentinewallace

CONTRIBUTING.md

@@ -192,6 +192,12 @@ welcomed.
 
 * `LDK_TEST_DETERMINISTIC_HASHES` - When set to `1`, uses deterministic hash map iteration order in tests. This ensures consistent test output across runs, useful for comparing logs before and after changes.
 
+* `LDK_TEST_REBUILD_MGR_FROM_MONITORS` - If set to `1`, on test node reload the `ChannelManager`'s
+  HTLC set will be reconstructed from `Channel{Monitor}` persisted data. If `0`, test nodes will be
+  reloaded from persisted `ChannelManager` data using legacy code paths. This ensures consistent
+  test output across runs, useful for comparing logs before and after changes, since otherwise the
+  selection of which codepaths to be used on reload will be chosen randomly.
+
 C/C++ Bindings
 --------------
 

lightning/src/ln/channelmanager.rs

@@ -11747,6 +11747,10 @@ 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();
+				// It's possible we processed this intercept forward, generated an event, then re-processed
+				// it here after restart, in which case the intercept event should not be pushed
+				// redundantly.
+				new_intercept_events.retain(|ev| !events.contains(ev));
 				events.append(&mut new_intercept_events);
 			}
 		}
@@ -17484,9 +17488,9 @@ where
 
 		const MAX_ALLOC_SIZE: usize = 1024 * 64;
 		let forward_htlcs_count: u64 = Readable::read(reader)?;
-		// This map is read but may no longer be used because we'll attempt to rebuild the set of HTLC
-		// forwards from the `Channel{Monitor}`s instead, as a step towards removing the requirement of
-		// regularly persisting the `ChannelManager`.
+		// Marked `_legacy` because in versions > 0.2 we are taking steps to remove the requirement of
+		// regularly persisting the `ChannelManager` and instead rebuild the set of HTLC forwards from
+		// `Channel{Monitor}` data. See `reconstruct_manager_from_monitors` usage below.
 		let mut forward_htlcs_legacy: HashMap<u64, Vec<HTLCForwardInfo>> =
 			hash_map_with_capacity(cmp::min(forward_htlcs_count as usize, 128));
 		for _ in 0..forward_htlcs_count {
@@ -17587,9 +17591,9 @@ where
 			};
 		}
 
-		// Some maps are read but may no longer be used because we attempt to rebuild the pending HTLC
-		// set from the `Channel{Monitor}`s instead, as a step towards removing the requirement of
-		// regularly persisting the `ChannelManager`.
+		// Marked `_legacy` because in versions > 0.2 we are taking steps to remove the requirement of
+		// regularly persisting the `ChannelManager` and instead rebuild the set of HTLC forwards from
+		// `Channel{Monitor}` data. See `reconstruct_manager_from_monitors` below.
 		let mut pending_intercepted_htlcs_legacy: Option<HashMap<InterceptId, PendingAddHTLCInfo>> =
 			None;
 		let mut decode_update_add_htlcs_legacy: Option<HashMap<u64, Vec<msgs::UpdateAddHTLC>>> =
@@ -17930,6 +17934,36 @@ where
 			pending_background_events.push(new_event);
 		}
 
+		// In LDK 0.2 and below, the `ChannelManager` would track all payments and HTLCs internally and
+		// persist that state, relying on it being up-to-date on restart. Newer versions are moving
+		// towards reducing this reliance on regular persistence of the `ChannelManager`, and instead
+		// reconstruct HTLC/payment state based on `Channel{Monitor}` data if
+		// `reconstruct_manager_from_monitors` is set below. Currently it is only set in tests, randomly
+		// to ensure the legacy codepaths also have test coverage.
+		#[cfg(not(test))]
+		let reconstruct_manager_from_monitors = false;
+		#[cfg(test)]
+		let reconstruct_manager_from_monitors = {
+			use core::hash::{BuildHasher, Hasher};
+
+			match std::env::var("LDK_TEST_REBUILD_MGR_FROM_MONITORS") {
+				Ok(val) => match val.as_str() {
+					"1" => true,
+					"0" => false,
+					_ => panic!("LDK_TEST_REBUILD_MGR_FROM_MONITORS must be 0 or 1, got: {}", val),
+				},
+				Err(_) => {
+					let rand_val =
+						std::collections::hash_map::RandomState::new().build_hasher().finish();
+					if rand_val % 2 == 0 {
+						true
+					} else {
+						false
+					}
+				},
+			}
+		};
+
 		// If there's any preimages for forwarded HTLCs hanging around in ChannelMonitors we
 		// should ensure we try them again on the inbound edge. We put them here and do so after we
 		// have a fully-constructed `ChannelManager` at the end.
@@ -17954,18 +17988,20 @@ where
 					let mut peer_state_lock = peer_state_mtx.lock().unwrap();
 					let peer_state = &mut *peer_state_lock;
 					is_channel_closed = !peer_state.channel_by_id.contains_key(channel_id);
-					if let Some(chan) = peer_state.channel_by_id.get(channel_id) {
-						if let Some(funded_chan) = chan.as_funded() {
-							let inbound_committed_update_adds =
-								funded_chan.get_inbound_committed_update_adds();
-							if !inbound_committed_update_adds.is_empty() {
-								// Reconstruct `ChannelManager::decode_update_add_htlcs` from the serialized
-								// `Channel`, as part of removing the requirement to regularly persist the
-								// `ChannelManager`.
-								decode_update_add_htlcs.insert(
-									funded_chan.context.outbound_scid_alias(),
-									inbound_committed_update_adds,
-								);
+					if reconstruct_manager_from_monitors {
+						if let Some(chan) = peer_state.channel_by_id.get(channel_id) {
+							if let Some(funded_chan) = chan.as_funded() {
+								let inbound_committed_update_adds =
+									funded_chan.get_inbound_committed_update_adds();
+								if !inbound_committed_update_adds.is_empty() {
+									// Reconstruct `ChannelManager::decode_update_add_htlcs` from the serialized
+									// `Channel`, as part of removing the requirement to regularly persist the
+									// `ChannelManager`.
+									decode_update_add_htlcs.insert(
+										funded_chan.context.outbound_scid_alias(),
+										inbound_committed_update_adds,
+									);
+								}
 							}
 						}
 					}
@@ -18020,17 +18056,20 @@ where
 								info.prev_funding_outpoint == prev_hop_data.outpoint
 									&& info.prev_htlc_id == prev_hop_data.htlc_id
 							};
-							// We always add all inbound committed HTLCs to `decode_update_add_htlcs` in the above
-							// loop, but we need to prune from those added HTLCs if they were already forwarded to
-							// the outbound edge. Otherwise, we'll double-forward.
-							dedup_decode_update_add_htlcs(
-								&mut decode_update_add_htlcs,
-								&prev_hop_data,
-								"HTLC was forwarded to the closed channel",
-								&args.logger,
-							);
+							// If `reconstruct_manager_from_monitors` is set, we always add all inbound committed
+							// HTLCs to `decode_update_add_htlcs` in the above loop, but we need to prune from
+							// those added HTLCs if they were already forwarded to the outbound edge. Otherwise,
+							// we'll double-forward.
+							if reconstruct_manager_from_monitors {
+								dedup_decode_update_add_htlcs(
+									&mut decode_update_add_htlcs,
+									&prev_hop_data,
+									"HTLC was forwarded to the closed channel",
+									&args.logger,
+								);
+							}
 
-							if !is_channel_closed {
+							if !is_channel_closed || reconstruct_manager_from_monitors {
 								continue;
 							}
 							// The ChannelMonitor is now responsible for this HTLC's
@@ -18539,99 +18578,55 @@ where
 			}
 		}
 
-		// De-duplicate HTLCs that are present in both `failed_htlcs` and `decode_update_add_htlcs`.
-		// Omitting this de-duplication could lead to redundant HTLC processing and/or bugs.
-		for (src, _, _, _, _, _) in failed_htlcs.iter() {
-			if let HTLCSource::PreviousHopData(prev_hop_data) = src {
-				dedup_decode_update_add_htlcs(
-					&mut decode_update_add_htlcs,
-					prev_hop_data,
-					"HTLC was failed backwards during manager read",
-					&args.logger,
-				);
-			}
-		}
-
-		// See above comment on `failed_htlcs`.
-		for htlcs in claimable_payments.values().map(|pmt| &pmt.htlcs) {
-			for prev_hop_data in htlcs.iter().map(|h| &h.prev_hop) {
-				dedup_decode_update_add_htlcs(
-					&mut decode_update_add_htlcs,
-					prev_hop_data,
-					"HTLC was already decoded and marked as a claimable payment",
-					&args.logger,
-				);
-			}
-		}
-
-		// 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;
-					}
+		if reconstruct_manager_from_monitors {
+			// De-duplicate HTLCs that are present in both `failed_htlcs` and `decode_update_add_htlcs`.
+			// Omitting this de-duplication could lead to redundant HTLC processing and/or bugs.
+			for (src, _, _, _, _, _) in failed_htlcs.iter() {
+				if let HTLCSource::PreviousHopData(prev_hop_data) = src {
+					dedup_decode_update_add_htlcs(
+						&mut decode_update_add_htlcs,
+						prev_hop_data,
+						"HTLC was failed backwards during manager read",
+						&args.logger,
+					);
 				}
 			}
-			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),
+
+			// See above comment on `failed_htlcs`.
+			for htlcs in claimable_payments.values().map(|pmt| &pmt.htlcs) {
+				for prev_hop_data in htlcs.iter().map(|h| &h.prev_hop) {
+					dedup_decode_update_add_htlcs(
+						&mut decode_update_add_htlcs,
+						prev_hop_data,
+						"HTLC was already decoded and marked as a claimable payment",
+						&args.logger,
 					);
-					return false;
 				}
 			}
+		}
+
+		let (decode_update_add_htlcs, forward_htlcs, pending_intercepted_htlcs) =
+			if reconstruct_manager_from_monitors {
+				(decode_update_add_htlcs, new_hash_map(), new_hash_map())
+			} else {
+				(
+					decode_update_add_htlcs_legacy,
+					forward_htlcs_legacy,
+					pending_intercepted_htlcs_legacy,
+				)
+			};
+
+		// If we have a pending intercept HTLC present but no corresponding event, add that now rather
+		// than relying on the user having persisted the event prior to shutdown.
+		for (id, fwd) in pending_intercepted_htlcs.iter() {
 			if !pending_events_read.iter().any(
 				|(ev, _)| matches!(ev, Event::HTLCIntercepted { intercept_id, .. } if intercept_id == id),
 			) {
-				match create_htlc_intercepted_event(*id, &fwd) {
+				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);
@@ -18660,9 +18655,9 @@ where
 
 			inbound_payment_key: expanded_inbound_key,
 			pending_outbound_payments: pending_outbounds,
-			pending_intercepted_htlcs: Mutex::new(pending_intercepted_htlcs_legacy),
+			pending_intercepted_htlcs: Mutex::new(pending_intercepted_htlcs),
 
-			forward_htlcs: Mutex::new(forward_htlcs_legacy),
+			forward_htlcs: Mutex::new(forward_htlcs),
 			decode_update_add_htlcs: Mutex::new(decode_update_add_htlcs),
 			claimable_payments: Mutex::new(ClaimablePayments {
 				claimable_payments,
@@ -18998,12 +18993,11 @@ where
 mod tests {
 	use crate::events::{ClosureReason, Event, HTLCHandlingFailureType};
 	use crate::ln::channelmanager::{
-		create_recv_pending_htlc_info, inbound_payment, HTLCForwardInfo, InterceptId, PaymentId,
+		create_recv_pending_htlc_info, inbound_payment, InterceptId, PaymentId,
 		RecipientOnionFields,
 	};
 	use crate::ln::functional_test_utils::*;
 	use crate::ln::msgs::{self, BaseMessageHandler, ChannelMessageHandler, MessageSendEvent};
-	use crate::ln::onion_utils::AttributionData;
 	use crate::ln::onion_utils::{self, LocalHTLCFailureReason};
 	use crate::ln::outbound_payment::Retry;
 	use crate::ln::types::ChannelId;
@@ -19013,7 +19007,6 @@ mod tests {
 	use crate::types::payment::{PaymentHash, PaymentPreimage, PaymentSecret};
 	use crate::util::config::{ChannelConfig, ChannelConfigUpdate};
 	use crate::util::errors::APIError;
-	use crate::util::ser::Writeable;
 	use crate::util::test_utils;
 	use bitcoin::secp256k1::ecdh::SharedSecret;
 	use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
@@ -20071,66 +20064,6 @@ mod tests {
 			check_spends!(txn[0], funding_tx);
 		}
 	}
-
-	#[test]
-	#[rustfmt::skip]
-	fn test_malformed_forward_htlcs_ser() {
-		// Ensure that `HTLCForwardInfo::FailMalformedHTLC`s are (de)serialized properly.
-		let chanmon_cfg = create_chanmon_cfgs(1);
-		let node_cfg = create_node_cfgs(1, &chanmon_cfg);
-		let persister;
-		let chain_monitor;
-		let chanmgrs = create_node_chanmgrs(1, &node_cfg, &[None]);
-		let deserialized_chanmgr;
-		let mut nodes = create_network(1, &node_cfg, &chanmgrs);
-
-		let dummy_failed_htlc = |htlc_id| {
-			HTLCForwardInfo::FailHTLC { htlc_id, err_packet: msgs::OnionErrorPacket { data: vec![42], attribution_data: Some(AttributionData::new()) } }
-		};
-		let dummy_malformed_htlc = |htlc_id| {
-			HTLCForwardInfo::FailMalformedHTLC {
-				htlc_id,
-				failure_code: LocalHTLCFailureReason::InvalidOnionPayload.failure_code(),
-				sha256_of_onion: [0; 32],
-			}
-		};
-
-		let dummy_htlcs_1: Vec<HTLCForwardInfo> = (1..10).map(|htlc_id| {
-			if htlc_id % 2 == 0 {
-				dummy_failed_htlc(htlc_id)
-			} else {
-				dummy_malformed_htlc(htlc_id)
-			}
-		}).collect();
-
-		let dummy_htlcs_2: Vec<HTLCForwardInfo> = (1..10).map(|htlc_id| {
-			if htlc_id % 2 == 1 {
-				dummy_failed_htlc(htlc_id)
-			} else {
-				dummy_malformed_htlc(htlc_id)
-			}
-		}).collect();
-
-
-		let (scid_1, scid_2) = (42, 43);
-		let mut forward_htlcs = new_hash_map();
-		forward_htlcs.insert(scid_1, dummy_htlcs_1.clone());
-		forward_htlcs.insert(scid_2, dummy_htlcs_2.clone());
-
-		let mut chanmgr_fwd_htlcs = nodes[0].node.forward_htlcs.lock().unwrap();
-		*chanmgr_fwd_htlcs = forward_htlcs.clone();
-		core::mem::drop(chanmgr_fwd_htlcs);
-
-		reload_node!(nodes[0], nodes[0].node.encode(), &[], persister, chain_monitor, deserialized_chanmgr);
-
-		let mut deserialized_fwd_htlcs = nodes[0].node.forward_htlcs.lock().unwrap();
-		for scid in [scid_1, scid_2].iter() {
-			let deserialized_htlcs = deserialized_fwd_htlcs.remove(scid).unwrap();
-			assert_eq!(forward_htlcs.remove(scid).unwrap(), deserialized_htlcs);
-		}
-		assert!(deserialized_fwd_htlcs.is_empty());
-		core::mem::drop(deserialized_fwd_htlcs);
-	}
 }
 
 #[cfg(ldk_bench)]