Move receiving HTLCs to a receive_htlcs field

Previously, we'd store receiving HTLCs side-by-side with HTLCs forwards
in the `forwards_htlcs` field under SCID 0.

Here, we opt to split out a separate `receive_htlcs` field, also
omitting the 0 magic value.

Author: tnull

lightning/src/ln/channelmanager.rs

@@ -2462,6 +2462,8 @@ where
 //  |   |
 //  |   |__`pending_intercepted_htlcs`
 //  |
+//  |__`receive_htlcs`
+//  |
 //  |__`decode_update_add_htlcs`
 //  |
 //  |__`per_peer_state`
@@ -2537,7 +2539,7 @@ pub struct ChannelManager<
 	/// See `ChannelManager` struct-level documentation for lock order requirements.
 	pending_outbound_payments: OutboundPayments,
 
-	/// SCID/SCID Alias -> forward infos. Key of 0 means payments received.
+	/// SCID/SCID Alias -> forward infos.
 	///
 	/// Note that because we may have an SCID Alias as the key we can have two entries per channel,
 	/// though in practice we probably won't be receiving HTLCs for a channel both via the alias
@@ -2556,7 +2558,13 @@ pub struct ChannelManager<
 	///
 	/// See `ChannelManager` struct-level documentation for lock order requirements.
 	pending_intercepted_htlcs: Mutex<HashMap<InterceptId, PendingAddHTLCInfo>>,
-
+	/// Storage for HTLCs that are meant for us.
+	///
+	/// See `ChannelManager` struct-level documentation for lock order requirements.
+	#[cfg(test)]
+	pub(super) receive_htlcs: Mutex<Vec<HTLCForwardInfo>>,
+	#[cfg(not(test))]
+	receive_htlcs: Mutex<Vec<HTLCForwardInfo>>,
 	/// SCID/SCID Alias -> pending `update_add_htlc`s to decode.
 	///
 	/// Note that because we may have an SCID Alias as the key we can have two entries per channel,
@@ -3738,6 +3746,7 @@ where
 			outbound_scid_aliases: Mutex::new(new_hash_set()),
 			pending_outbound_payments: OutboundPayments::new(new_hash_map()),
 			forward_htlcs: Mutex::new(new_hash_map()),
+			receive_htlcs: Mutex::new(Vec::new()),
 			decode_update_add_htlcs: Mutex::new(new_hash_map()),
 			claimable_payments: Mutex::new(ClaimablePayments { claimable_payments: new_hash_map(), pending_claiming_payments: new_hash_map() }),
 			pending_intercepted_htlcs: Mutex::new(new_hash_map()),
@@ -6355,6 +6364,9 @@ where
 		if !self.forward_htlcs.lock().unwrap().is_empty() {
 			return true;
 		}
+		if !self.receive_htlcs.lock().unwrap().is_empty() {
+			return true;
+		}
 		if !self.decode_update_add_htlcs.lock().unwrap().is_empty() {
 			return true;
 		}
@@ -6402,22 +6414,18 @@ where
 
 		for (short_chan_id, mut pending_forwards) in forward_htlcs {
 			should_persist = NotifyOption::DoPersist;
-			if short_chan_id != 0 {
-				self.process_forward_htlcs(
-					short_chan_id,
-					&mut pending_forwards,
-					&mut failed_forwards,
-					&mut phantom_receives,
-				);
-			} else {
-				self.process_receive_htlcs(
-					&mut pending_forwards,
-					&mut new_events,
-					&mut failed_forwards,
-				);
-			}
+			self.process_forward_htlcs(
+				short_chan_id,
+				&mut pending_forwards,
+				&mut failed_forwards,
+				&mut phantom_receives,
+			);
 		}
 
+		let mut receive_htlcs = Vec::new();
+		mem::swap(&mut receive_htlcs, &mut self.receive_htlcs.lock().unwrap());
+		self.process_receive_htlcs(receive_htlcs, &mut new_events, &mut failed_forwards);
+
 		let best_block_height = self.best_block.read().unwrap().height;
 		let needs_persist = self.pending_outbound_payments.check_retry_payments(
 			&self.router,
@@ -6929,11 +6937,11 @@ where
 	}
 
 	fn process_receive_htlcs(
-		&self, pending_forwards: &mut Vec<HTLCForwardInfo>,
+		&self, receive_htlcs: Vec<HTLCForwardInfo>,
 		new_events: &mut VecDeque<(Event, Option<EventCompletionAction>)>,
 		failed_forwards: &mut Vec<FailedHTLCForward>,
 	) {
-		'next_forwardable_htlc: for forward_info in pending_forwards.drain(..) {
+		'next_forwardable_htlc: for forward_info in receive_htlcs.into_iter() {
 			match forward_info {
 				HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
 					prev_short_channel_id,
@@ -10346,8 +10354,21 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					let scid = match forward_info.routing {
 						PendingHTLCRouting::Forward { short_channel_id, .. } => short_channel_id,
 						PendingHTLCRouting::TrampolineForward { .. } => 0,
-						PendingHTLCRouting::Receive { .. } => 0,
-						PendingHTLCRouting::ReceiveKeysend { .. } => 0,
+						PendingHTLCRouting::Receive { .. }
+						| PendingHTLCRouting::ReceiveKeysend { .. } => {
+							self.receive_htlcs.lock().unwrap().push(HTLCForwardInfo::AddHTLC(
+								PendingAddHTLCInfo {
+									prev_short_channel_id,
+									prev_counterparty_node_id,
+									prev_funding_outpoint,
+									prev_channel_id,
+									prev_htlc_id,
+									prev_user_channel_id,
+									forward_info,
+								},
+							));
+							continue;
+						},
 					};
 					// Pull this now to avoid introducing a lock order with `forward_htlcs`.
 					let is_our_scid = self.short_to_chan_info.read().unwrap().contains_key(&scid);
@@ -15091,6 +15112,8 @@ where
 			}
 		}
 
+		let receive_htlcs = self.receive_htlcs.lock().unwrap();
+
 		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() {
@@ -15258,6 +15281,7 @@ where
 			(17, in_flight_monitor_updates, option),
 			(19, peer_storage_dir, optional_vec),
 			(21, self.flow.writeable_async_receive_offer_cache(), required),
+			(23, *receive_htlcs, required_vec),
 		});
 
 		Ok(())
@@ -15818,6 +15842,7 @@ where
 		const MAX_ALLOC_SIZE: usize = 1024 * 64;
 		let forward_htlcs_count: u64 = Readable::read(reader)?;
 		let mut forward_htlcs = hash_map_with_capacity(cmp::min(forward_htlcs_count as usize, 128));
+		let mut legacy_receive_htlcs: Vec<HTLCForwardInfo> = Vec::new();
 		for _ in 0..forward_htlcs_count {
 			let short_channel_id = Readable::read(reader)?;
 			let pending_forwards_count: u64 = Readable::read(reader)?;
@@ -15826,7 +15851,26 @@ where
 				MAX_ALLOC_SIZE / mem::size_of::<HTLCForwardInfo>(),
 			));
 			for _ in 0..pending_forwards_count {
-				pending_forwards.push(Readable::read(reader)?);
+				let pending_htlc = Readable::read(reader)?;
+				// Prior to LDK 0.2, Receive HTLCs used to be stored in `forward_htlcs` under SCID == 0. Here we migrate
+				// the old data if necessary.
+				if short_channel_id == 0 {
+					match pending_htlc {
+						HTLCForwardInfo::AddHTLC(ref htlc_info) => {
+							if matches!(
+								htlc_info.forward_info.routing,
+								PendingHTLCRouting::Receive { .. }
+									| PendingHTLCRouting::ReceiveKeysend { .. }
+							) {
+								legacy_receive_htlcs.push(pending_htlc);
+								continue;
+							}
+						},
+						_ => {},
+					}
+				}
+
+				pending_forwards.push(pending_htlc);
 			}
 			forward_htlcs.insert(short_channel_id, pending_forwards);
 		}
@@ -15943,6 +15987,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 receive_htlcs = None;
 		read_tlv_fields!(reader, {
 			(1, pending_outbound_payments_no_retry, option),
 			(2, pending_intercepted_htlcs, option),
@@ -15961,8 +16006,10 @@ 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, receive_htlcs, optional_vec),
 		});
 		let mut decode_update_add_htlcs = decode_update_add_htlcs.unwrap_or_else(|| new_hash_map());
+		let receive_htlcs = receive_htlcs.unwrap_or_else(|| legacy_receive_htlcs);
 		let peer_storage_dir: Vec<(PublicKey, Vec<u8>)> = peer_storage_dir.unwrap_or_else(Vec::new);
 		if fake_scid_rand_bytes.is_none() {
 			fake_scid_rand_bytes = Some(args.entropy_source.get_secure_random_bytes());
@@ -16791,6 +16838,7 @@ where
 			pending_intercepted_htlcs: Mutex::new(pending_intercepted_htlcs.unwrap()),
 
 			forward_htlcs: Mutex::new(forward_htlcs),
+			receive_htlcs: Mutex::new(receive_htlcs),
 			decode_update_add_htlcs: Mutex::new(decode_update_add_htlcs),
 			claimable_payments: Mutex::new(ClaimablePayments {
 				claimable_payments,

lightning/src/ln/payment_tests.rs

@@ -636,30 +636,23 @@ fn test_reject_mpp_keysend_htlc_mismatching_secret() {
 	nodes[3].node.process_pending_update_add_htlcs();
 
 	assert!(nodes[3].node.get_and_clear_pending_msg_events().is_empty());
-	assert_eq!(nodes[3].node.forward_htlcs.lock().unwrap().len(), 1);
-	if let Some((_, pending_forwards)) =
-		nodes[3].node.forward_htlcs.lock().unwrap().iter_mut().next()
-	{
-		assert_eq!(pending_forwards.len(), 1);
-		match pending_forwards.get_mut(0).unwrap() {
-			&mut HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo { ref mut forward_info, .. }) => {
-				match forward_info.routing {
-					PendingHTLCRouting::ReceiveKeysend { ref mut payment_data, .. } => {
-						*payment_data = Some(msgs::FinalOnionHopData {
-							payment_secret: PaymentSecret([42; 32]),
-							total_msat: amount * 2,
-						});
-					},
-					_ => panic!("Expected PendingHTLCRouting::ReceiveKeysend"),
-				}
-			},
-			_ => {
-				panic!("Unexpected HTLCForwardInfo");
-			},
-		}
-	} else {
-		panic!("Expected pending receive");
-	};
+	assert_eq!(nodes[3].node.receive_htlcs.lock().unwrap().len(), 1);
+	match nodes[3].node.receive_htlcs.lock().unwrap().get_mut(0).unwrap() {
+		&mut HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo { ref mut forward_info, .. }) => {
+			match forward_info.routing {
+				PendingHTLCRouting::ReceiveKeysend { ref mut payment_data, .. } => {
+					*payment_data = Some(msgs::FinalOnionHopData {
+						payment_secret: PaymentSecret([42; 32]),
+						total_msat: amount * 2,
+					});
+				},
+				_ => panic!("Expected PendingHTLCRouting::ReceiveKeysend"),
+			}
+		},
+		_ => {
+			panic!("Unexpected HTLCForwardInfo");
+		},
+	}
 	nodes[3].node.process_pending_htlc_forwards();
 
 	// Pay along nodes[2]
@@ -687,34 +680,26 @@ fn test_reject_mpp_keysend_htlc_mismatching_secret() {
 	let update_add_3 = update_3.update_add_htlcs[0].clone();
 	nodes[3].node.handle_update_add_htlc(node_c_id, &update_add_3);
 	commitment_signed_dance!(nodes[3], nodes[2], update_3.commitment_signed, false, true);
-	expect_htlc_failure_conditions(nodes[3].node.get_and_clear_pending_events(), &[]);
-	nodes[3].node.process_pending_update_add_htlcs();
-
+	assert!(nodes[3].node.get_and_clear_pending_events().is_empty());
 	assert!(nodes[3].node.get_and_clear_pending_msg_events().is_empty());
-	assert_eq!(nodes[3].node.forward_htlcs.lock().unwrap().len(), 1);
-	if let Some((_, pending_forwards)) =
-		nodes[3].node.forward_htlcs.lock().unwrap().iter_mut().next()
-	{
-		assert_eq!(pending_forwards.len(), 1);
-		match pending_forwards.get_mut(0).unwrap() {
-			&mut HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo { ref mut forward_info, .. }) => {
-				match forward_info.routing {
-					PendingHTLCRouting::ReceiveKeysend { ref mut payment_data, .. } => {
-						*payment_data = Some(msgs::FinalOnionHopData {
-							payment_secret: PaymentSecret([43; 32]), // Doesn't match the secret used above
-							total_msat: amount * 2,
-						});
-					},
-					_ => panic!("Expected PendingHTLCRouting::ReceiveKeysend"),
-				}
-			},
-			_ => {
-				panic!("Unexpected HTLCForwardInfo");
-			},
-		}
-	} else {
-		panic!("Expected pending receive");
-	};
+	nodes[3].node.process_pending_update_add_htlcs();
+	assert_eq!(nodes[3].node.receive_htlcs.lock().unwrap().len(), 1);
+	match nodes[3].node.receive_htlcs.lock().unwrap().get_mut(0).unwrap() {
+		&mut HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo { ref mut forward_info, .. }) => {
+			match forward_info.routing {
+				PendingHTLCRouting::ReceiveKeysend { ref mut payment_data, .. } => {
+					*payment_data = Some(msgs::FinalOnionHopData {
+						payment_secret: PaymentSecret([43; 32]), // Doesn't match the secret used above
+						total_msat: amount * 2,
+					});
+				},
+				_ => panic!("Expected PendingHTLCRouting::ReceiveKeysend"),
+			}
+		},
+		_ => {
+			panic!("Unexpected HTLCForwardInfo");
+		},
+	}
 	nodes[3].node.process_pending_htlc_forwards();
 	let fail_type = HTLCHandlingFailureType::Receive { payment_hash };
 	expect_and_process_pending_htlcs_and_htlc_handling_failed(&nodes[3], &[fail_type]);