Support deleting legacy forward map persistence in 0.5

In 0.3+, we are taking steps to remove the requirement of regularly persisting
the ChannelManager and instead rebuild the set of HTLC forwards (and the
manager generally) from Channel{Monitor} data.

We previously merged support for reconstructing the
ChannelManager::decode_update_add_htlcs map from channel data, using a new
HTLC onion field that will be present for inbound HTLCs received on 0.3+ only.
The plan is that in upcoming LDK versions, the manager will reconstruct this
map and the other forward/claimable/pending HTLC maps will automatically
repopulate themselves on the next call to process_pending_htlc_forwards.

As such, once we're in a future version that reconstructs the pending HTLC set,
we can stop persisting the legacy ChannelManager maps such as forward_htlcs,
pending_intercepted_htlcs since they will never be used.

For 0.3 to be compatible with this future version, in this commit we detect
that the manager was last written on a version of LDK that doesn't persist the
legacy maps. In that case, we don't try to read the old forwards map and run
the new reconstruction logic only.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -16631,6 +16631,17 @@ pub fn provided_init_features(config: &UserConfig) -> InitFeatures {
 const SERIALIZATION_VERSION: u8 = 1;
 const MIN_SERIALIZATION_VERSION: u8 = 1;
 
+// We plan to start writing this version in 0.5.
+//
+// LDK 0.5+ will reconstruct the set of pending HTLCs from `Channel{Monitor}` data that started
+// being written in 0.3, ignoring legacy `ChannelManager` HTLC maps on read and not writing them.
+// LDK 0.5+ will automatically fail to read if the pending HTLC set cannot be reconstructed, i.e.
+// if we were last written with pending HTLCs on 0.2- or if the new 0.3+ fields are missing.
+//
+// If 0.3 or 0.4 reads this manager version, it knows that the legacy maps were not written and
+// acts accordingly.
+const RECONSTRUCT_HTLCS_FROM_CHANS_VERSION: u8 = 5;
+
 impl_writeable_tlv_based!(PhantomRouteHints, {
 	(2, channels, required_vec),
 	(4, phantom_scid, required),
@@ -17382,6 +17393,8 @@ pub(super) struct ChannelManagerData<SP: SignerProvider> {
 	forward_htlcs_legacy: HashMap<u64, Vec<HTLCForwardInfo>>,
 	pending_intercepted_htlcs_legacy: HashMap<InterceptId, PendingAddHTLCInfo>,
 	decode_update_add_htlcs_legacy: HashMap<u64, Vec<msgs::UpdateAddHTLC>>,
+	// The `ChannelManager` version that was written.
+	version: u8,
 }
 
 /// Arguments for deserializing [`ChannelManagerData`].
@@ -17405,7 +17418,7 @@ impl<'a, ES: EntropySource, NS: NodeSigner, SP: SignerProvider, L: Logger>
 	fn read<R: io::Read>(
 		reader: &mut R, args: ChannelManagerDataReadArgs<'a, ES, NS, SP, L>,
 	) -> Result<Self, DecodeError> {
-		let _ver = read_ver_prefix!(reader, SERIALIZATION_VERSION);
+		let version = read_ver_prefix!(reader, SERIALIZATION_VERSION);
 
 		let chain_hash: ChainHash = Readable::read(reader)?;
 		let best_block_height: u32 = Readable::read(reader)?;
@@ -17427,21 +17440,26 @@ impl<'a, ES: EntropySource, NS: NodeSigner, SP: SignerProvider, L: Logger>
 			channels.push(channel);
 		}
 
-		let forward_htlcs_count: u64 = Readable::read(reader)?;
-		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 {
-			let short_channel_id = Readable::read(reader)?;
-			let pending_forwards_count: u64 = Readable::read(reader)?;
-			let mut pending_forwards = Vec::with_capacity(cmp::min(
-				pending_forwards_count as usize,
-				MAX_ALLOC_SIZE / mem::size_of::<HTLCForwardInfo>(),
-			));
-			for _ in 0..pending_forwards_count {
-				pending_forwards.push(Readable::read(reader)?);
-			}
-			forward_htlcs_legacy.insert(short_channel_id, pending_forwards);
-		}
+		let forward_htlcs_legacy: HashMap<u64, Vec<HTLCForwardInfo>> =
+			if version < RECONSTRUCT_HTLCS_FROM_CHANS_VERSION {
+				let forward_htlcs_count: u64 = Readable::read(reader)?;
+				let mut fwds = hash_map_with_capacity(cmp::min(forward_htlcs_count as usize, 128));
+				for _ in 0..forward_htlcs_count {
+					let short_channel_id = Readable::read(reader)?;
+					let pending_forwards_count: u64 = Readable::read(reader)?;
+					let mut pending_forwards = Vec::with_capacity(cmp::min(
+						pending_forwards_count as usize,
+						MAX_ALLOC_SIZE / mem::size_of::<HTLCForwardInfo>(),
+					));
+					for _ in 0..pending_forwards_count {
+						pending_forwards.push(Readable::read(reader)?);
+					}
+					fwds.insert(short_channel_id, pending_forwards);
+				}
+				fwds
+			} else {
+				new_hash_map()
+			};
 
 		let claimable_htlcs_count: u64 = Readable::read(reader)?;
 		let mut claimable_htlcs_list =
@@ -17721,6 +17739,7 @@ impl<'a, ES: EntropySource, NS: NodeSigner, SP: SignerProvider, L: Logger>
 			in_flight_monitor_updates: in_flight_monitor_updates.unwrap_or_default(),
 			peer_storage_dir: peer_storage_dir.unwrap_or_default(),
 			async_receive_offer_cache,
+			version,
 		})
 	}
 }
@@ -18023,6 +18042,7 @@ impl<
 			mut in_flight_monitor_updates,
 			peer_storage_dir,
 			async_receive_offer_cache,
+			version: _version,
 		} = data;
 
 		let empty_peer_state = || PeerState {
@@ -18572,10 +18592,10 @@ impl<
 		// 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.
+		// `reconstruct_manager_from_monitors` is set below. Currently we set in tests randomly to
+		// ensure the legacy codepaths also have test coverage.
 		#[cfg(not(test))]
-		let reconstruct_manager_from_monitors = false;
+		let reconstruct_manager_from_monitors = _version >= RECONSTRUCT_HTLCS_FROM_CHANS_VERSION;
 		#[cfg(test)]
 		let reconstruct_manager_from_monitors =
 			args.reconstruct_manager_from_monitors.unwrap_or_else(|| {