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, prefer to use the old maps if they are present (which will always be
the case in prod, for now), but avoid writing the legacy maps in test mode so
tests will always exercise the new paths.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -11758,6 +11758,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);
 			}
 		}
@@ -16681,7 +16685,18 @@ where
 			}
 		}
 
-		{
+		// 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 on read. In tests, we want to always use the new
+		// codepaths.
+		#[cfg(not(any(test, feature = "_test_utils")))]
+		let reconstruct_manager_from_monitors = false;
+		#[cfg(any(test, feature = "_test_utils"))]
+		let reconstruct_manager_from_monitors = true;
+
+		if !reconstruct_manager_from_monitors {
 			let forward_htlcs = self.forward_htlcs.lock().unwrap();
 			(forward_htlcs.len() as u64).write(writer)?;
 			for (short_channel_id, pending_forwards) in forward_htlcs.iter() {
@@ -16691,12 +16706,16 @@ where
 					forward.write(writer)?;
 				}
 			}
+		} else {
+			0u64.write(writer)?;
 		}
 
 		let mut decode_update_add_htlcs_opt = None;
-		let decode_update_add_htlcs = self.decode_update_add_htlcs.lock().unwrap();
-		if !decode_update_add_htlcs.is_empty() {
-			decode_update_add_htlcs_opt = Some(decode_update_add_htlcs);
+		if !reconstruct_manager_from_monitors {
+			let decode_update_add_htlcs = self.decode_update_add_htlcs.lock().unwrap();
+			if !decode_update_add_htlcs.is_empty() {
+				decode_update_add_htlcs_opt = Some(decode_update_add_htlcs);
+			}
 		}
 
 		let claimable_payments = self.claimable_payments.lock().unwrap();
@@ -16842,9 +16861,11 @@ where
 		}
 
 		let mut pending_intercepted_htlcs = None;
-		let our_pending_intercepts = self.pending_intercepted_htlcs.lock().unwrap();
-		if our_pending_intercepts.len() != 0 {
-			pending_intercepted_htlcs = Some(our_pending_intercepts);
+		if !reconstruct_manager_from_monitors {
+			let our_pending_intercepts = self.pending_intercepted_htlcs.lock().unwrap();
+			if our_pending_intercepts.len() != 0 {
+				pending_intercepted_htlcs = Some(our_pending_intercepts);
+			}
 		}
 
 		let mut pending_claiming_payments = Some(&claimable_payments.pending_claiming_payments);
@@ -16885,6 +16906,7 @@ where
 			(17, in_flight_monitor_updates, option),
 			(19, peer_storage_dir, optional_vec),
 			(21, WithoutLength(&self.flow.writeable_async_receive_offer_cache()), required),
+			(23, reconstruct_manager_from_monitors, required),
 		});
 
 		// Remove the SpliceFailed events added earlier.
@@ -17597,9 +17619,10 @@ 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`.
+		// In LDK versions >0.2, we are taking steps to remove the requirement of regularly peristing
+		// the `ChannelManager`. To that end, if `reconstruct_manager_from_monitors` is set below, we
+		// will rebuild the pending HTLC set using data from the `Channel{Monitor}`s instead and ignore
+		// these legacy maps.
 		let mut pending_intercepted_htlcs_legacy: Option<HashMap<InterceptId, PendingAddHTLCInfo>> =
 			None;
 		let mut decode_update_add_htlcs_legacy: Option<HashMap<u64, Vec<msgs::UpdateAddHTLC>>> =
@@ -17629,6 +17652,7 @@ where
 		let mut inbound_payment_id_secret = None;
 		let mut peer_storage_dir: Option<Vec<(PublicKey, Vec<u8>)>> = None;
 		let mut async_receive_offer_cache: AsyncReceiveOfferCache = AsyncReceiveOfferCache::new();
+		let mut reconstruct_manager_from_monitors = false;
 		read_tlv_fields!(reader, {
 			(1, pending_outbound_payments_no_retry, option),
 			(2, pending_intercepted_htlcs_legacy, option),
@@ -17647,6 +17671,7 @@ where
 			(17, in_flight_monitor_updates, option),
 			(19, peer_storage_dir, optional_vec),
 			(21, async_receive_offer_cache, (default_value, async_receive_offer_cache)),
+			(23, reconstruct_manager_from_monitors, (default_value, false)),
 		});
 		let mut decode_update_add_htlcs_legacy =
 			decode_update_add_htlcs_legacy.unwrap_or_else(|| new_hash_map());
@@ -17964,18 +17989,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,
+									);
+								}
 							}
 						}
 					}
@@ -18030,15 +18057,18 @@ 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 {
 								// The ChannelMonitor is now responsible for this HTLC's
 								// failure/success and will let us know what its outcome is. If we
@@ -18547,101 +18577,49 @@ 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;
 				}
 			}
+		}
+
+		// 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, intercept) in pending_intercepted_htlcs_legacy.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, intercept) {
 					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);
-				},
-			}
 		}
 
+		if !reconstruct_manager_from_monitors {
+			decode_update_add_htlcs = decode_update_add_htlcs_legacy;
+		}
 		let best_block = BestBlock::new(best_block_hash, best_block_height);
 		let flow = OffersMessageFlow::new(
 			chain_hash,
@@ -19006,12 +18984,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;
@@ -19021,7 +18998,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};
@@ -20079,66 +20055,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)]