f - soft delete channels

Soft delete channels by marking them as closed. We don't want to fully
remove channels that have been force-closed and could still have pending
HTLCs to be resolved. We'll fully remove the channel if it's been marked
as closed and it does not have any pending HTLCs. Note that in the case
where a channel is removed and there was an HTLC where it was the
incoming, we'll return an error. I think this is ok and can be handled
by upstream (ChannelManager). If the incoming channel not found was
actually an invalid channel then we want to debug_assert on this. If the
incoming channel was a force-closed channel and we still had a monitor
that needed to be resolved, we can discard that channel not found error.

Author: elnosh

lightning/src/ln/resource_manager.rs

@@ -456,6 +456,11 @@ struct Channel {
 	/// Tracks which channels have misused the congestion bucket and the unix timestamp.
 	last_congestion_misuse: HashMap<u64, u64>,
 	protected_bucket: BucketResources,
+
+	/// Whether this channel has been closed. We keep channels around until all its
+	/// [`Self::pending_htlcs`] have been resolved. This ensures that HTLC resolutions that
+	/// arrive after a force-close can still update reputation and revenue correctly.
+	closed: bool,
 }
 
 impl Channel {
@@ -487,6 +492,7 @@ impl Channel {
 				bucket_allocations.protected_slots,
 				bucket_allocations.protected_liquidity,
 			),
+			closed: false,
 		})
 	}
 
@@ -685,31 +691,17 @@ impl DefaultResourceManager {
 	pub fn remove_channel(&self, channel_id: u64) -> Result<(), ()> {
 		let mut channels_lock = self.channels.lock().unwrap();
 
-		// Release bucket resources on each incoming channel for its pending HTLCs.
-		if let Some(removed_channel) = channels_lock.remove(&channel_id) {
-			for (htlc_ref, pending_htlc) in &removed_channel.pending_htlcs {
-				debug_assert!(channels_lock.get(&htlc_ref.incoming_channel_id).is_some());
-				if let Some(incoming_channel) = channels_lock.get_mut(&htlc_ref.incoming_channel_id)
-				{
-					let _ = match pending_htlc.bucket {
-						BucketAssigned::General => incoming_channel
-							.general_bucket
-							.remove_htlc(channel_id, pending_htlc.incoming_amount_msat),
-						BucketAssigned::Congestion => incoming_channel
-							.congestion_bucket
-							.remove_htlc(pending_htlc.incoming_amount_msat),
-						BucketAssigned::Protected => incoming_channel
-							.protected_bucket
-							.remove_htlc(pending_htlc.incoming_amount_msat),
-					};
-				}
+		// If the channel has pending HTLCs, it is soft-deleted and will be fully removed once
+		// all its pending HTLCs have been resolved.
+		let channel = channels_lock.get_mut(&channel_id).ok_or(())?;
+		if channel.pending_htlcs.is_empty() {
+			// If the channel had no pending HTLCs, we can remove it.
+			channels_lock.remove(&channel_id);
+			for (_, channel) in channels_lock.iter_mut() {
+				channel.general_bucket.remove_channel_slots(channel_id);
 			}
-		}
-
-		// Clean up pending HTLC entries and channel slots.
-		for (_, channel) in channels_lock.iter_mut() {
-			channel.pending_htlcs.retain(|htlc_ref, _| htlc_ref.incoming_channel_id != channel_id);
-			channel.general_bucket.remove_channel_slots(channel_id);
+		} else {
+			channel.closed = true;
 		}
 		Ok(())
 	}
@@ -736,7 +728,8 @@ impl DefaultResourceManager {
 		let mut channels_lock = self.channels.lock().unwrap();
 
 		let htlc_ref = HtlcRef { incoming_channel_id, htlc_id };
-		let outgoing_channel = channels_lock.get_mut(&outgoing_channel_id).ok_or(())?;
+		let outgoing_channel =
+			channels_lock.get_mut(&outgoing_channel_id).filter(|c| !c.closed).ok_or(())?;
 
 		if outgoing_channel.pending_htlcs.get(&htlc_ref).is_some() {
 			return Err(());
@@ -749,7 +742,8 @@ impl DefaultResourceManager {
 		let in_flight_htlc_risk = self.htlc_in_flight_risk(fee, incoming_cltv_expiry, height_added);
 		let pending_htlcs_in_congestion = outgoing_channel.pending_htlcs_in_congestion();
 
-		let incoming_channel = channels_lock.get_mut(&incoming_channel_id).ok_or(())?;
+		let incoming_channel =
+			channels_lock.get_mut(&incoming_channel_id).filter(|c| !c.closed).ok_or(())?;
 
 		let (accountable, bucket_assigned) = if !incoming_accountable {
 			if incoming_channel.general_available(
@@ -865,6 +859,8 @@ impl DefaultResourceManager {
 		// `last_updated_unix_secs` is frequently mutated. We accept the clamping rather
 		// than erroring, as rejecting out-of-order timestamps would leave resources stuck.
 		outgoing_channel.outgoing_reputation.add_value(effective_fee, resolved_at);
+		let should_remove_outgoing =
+			outgoing_channel.closed && outgoing_channel.pending_htlcs.is_empty();
 
 		let incoming_channel = channels_lock.get_mut(&incoming_channel_id).ok_or(())?;
 		match pending_htlc.bucket {
@@ -890,6 +886,15 @@ impl DefaultResourceManager {
 			incoming_channel.incoming_revenue.add_value(fee, resolved_at);
 		}
 
+		// If the channel had been marked as closed and it does not have any other pending
+		// HTLCs, it can be fully removed now.
+		if should_remove_outgoing {
+			channels_lock.remove(&outgoing_channel_id);
+			for (_, channel) in channels_lock.iter_mut() {
+				channel.general_bucket.remove_channel_slots(outgoing_channel_id);
+			}
+		}
+
 		Ok(())
 	}
 }
@@ -2291,6 +2296,93 @@ mod tests {
 		assert!(get_htlc_bucket(&rm, INCOMING_SCID, htlc_id, OUTGOING_SCID_2).is_none());
 	}
 
+	#[test]
+	fn test_remove_channel_no_pending_htlcs() {
+		let rm = create_test_resource_manager_with_channels();
+		let channels = rm.channels.lock().unwrap();
+		assert!(channels.get(&OUTGOING_SCID).is_some());
+		drop(channels);
+
+		rm.remove_channel(OUTGOING_SCID).unwrap();
+
+		let channels = rm.channels.lock().unwrap();
+		assert!(channels.get(&OUTGOING_SCID).is_none());
+	}
+
+	#[test]
+	fn test_resolve_htlc_after_outgoing_channel_close() {
+		let entropy_source = TestKeysInterface::new(&[0; 32], Network::Testnet);
+		let rm = create_test_resource_manager_with_channels();
+		let htlc_id = 1;
+
+		let added_at = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+		assert_eq!(
+			add_test_htlc(&rm, false, htlc_id, Some(added_at), &entropy_source).unwrap(),
+			ForwardingOutcome::Forward(false),
+		);
+
+		// If a channel gets closed while having pending HTLCs, it should not be fully removed.
+		// It should only get marked as closed and wait until all pending HTLCs are fully
+		// resolved to remove it.
+		rm.remove_channel(OUTGOING_SCID).unwrap();
+		{
+			let channels = rm.channels.lock().unwrap();
+			let outgoing = channels.get(&OUTGOING_SCID).unwrap();
+			assert!(outgoing.closed);
+			assert_eq!(outgoing.pending_htlcs.len(), 1);
+		}
+
+		// New HTLCs should be rejected on the closed channel.
+		assert!(add_test_htlc(&rm, false, 2, None, &entropy_source).is_err());
+
+		// Resolve pending HTLC where closed channel is involved and then verify that it was
+		// fully removed since it does not have any other pending HTLCs.
+		let resolved_at = added_at + 10;
+		rm.resolve_htlc(INCOMING_SCID, htlc_id, OUTGOING_SCID, true, resolved_at).unwrap();
+
+		let channels = rm.channels.lock().unwrap();
+		assert!(channels.get(&OUTGOING_SCID).is_none());
+
+		// Incoming channel should have revenue updated (HTLC was settled).
+		let incoming = channels.get(&INCOMING_SCID).unwrap();
+		assert_eq!(incoming.incoming_revenue.aggregated_decaying_average.value, FEE_AMOUNT as i64,);
+	}
+
+	#[test]
+	fn test_resolve_htlc_after_incoming_channel_close() {
+		let entropy_source = TestKeysInterface::new(&[0; 32], Network::Testnet);
+		let rm = create_test_resource_manager_with_channels();
+		let htlc_id = 1;
+
+		let added_at = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+		assert_eq!(
+			add_test_htlc(&rm, false, htlc_id, Some(added_at), &entropy_source).unwrap(),
+			ForwardingOutcome::Forward(false),
+		);
+
+		// Close the incoming channel. It has no pending HTLCs as outgoing so it is
+		// hard-removed immediately. The HTLC on the outgoing channel still references
+		// it as the incoming link.
+		rm.remove_channel(INCOMING_SCID).unwrap();
+		{
+			let channels = rm.channels.lock().unwrap();
+			assert!(channels.get(&INCOMING_SCID).is_none());
+
+			// The pending HTLC on the outgoing channel is still present.
+			let outgoing = channels.get(&OUTGOING_SCID).unwrap();
+			assert_eq!(outgoing.pending_htlcs.len(), 1);
+		}
+
+		// Resolve fails because the incoming channel is gone but outgoing reputation is still
+		// updated on the outgoing channel.
+		let resolved_at = added_at + 10;
+		assert!(rm.resolve_htlc(INCOMING_SCID, htlc_id, OUTGOING_SCID, true, resolved_at).is_err());
+
+		let channels = rm.channels.lock().unwrap();
+		let outgoing = channels.get(&OUTGOING_SCID).unwrap();
+		assert_eq!(outgoing.outgoing_reputation.value, FEE_AMOUNT as i64);
+	}
+
 	#[test]
 	fn test_decaying_average_bounds() {
 		for (start, bound) in [(1000, i64::MAX), (-1000, i64::MIN)] {