Prune closed LSPS2 terminal channel state

Terminal JIT channel state is only useful while the forwarded channel
still exists. Drop completed LSPS2 mappings once the channel is gone so
persisted service state does not retain stale entries indefinitely.

Co-Authored-By: HAL 9000

Author: tnull

lightning-liquidity/src/lsps2/service.rs

@@ -644,6 +644,33 @@ impl PeerState {
 		});
 	}
 
+	fn prune_terminal_state_without_channels(
+		&mut self, live_channel_ids: &[ChannelId],
+	) -> Vec<(u64, ChannelId)> {
+		let mut pruned_channels = Vec::new();
+		self.outbound_channels_by_intercept_scid.retain(|intercept_scid, entry| {
+			if let Some(channel_id) = entry.get_channel_id() {
+				if !live_channel_ids.contains(&channel_id) {
+					pruned_channels.push((*intercept_scid, channel_id));
+					return false;
+				}
+			}
+			true
+		});
+
+		if pruned_channels.is_empty() {
+			return pruned_channels;
+		}
+
+		for (intercept_scid, channel_id) in pruned_channels.iter() {
+			self.intercept_scid_by_channel_id.remove(channel_id);
+			self.intercept_scid_by_user_channel_id.retain(|_, iscid| intercept_scid != iscid);
+		}
+		self.needs_persist = true;
+
+		pruned_channels
+	}
+
 	fn pending_requests_and_channels(&self) -> usize {
 		let pending_requests = self.pending_requests.len();
 		let pending_outbound_channels = self
@@ -1805,6 +1832,13 @@ where
 		loop {
 			let mut need_remove = Vec::new();
 			let mut need_persist = Vec::new();
+			let live_channel_ids: Vec<ChannelId> = self
+				.channel_manager
+				.get_cm()
+				.list_channels()
+				.into_iter()
+				.map(|details| details.channel_id)
+				.collect();
 
 			{
 				// First build a list of peers to persist and prune with the read lock. This allows
@@ -1813,22 +1847,43 @@ where
 				for (counterparty_node_id, inner_state_lock) in outer_state_lock.iter() {
 					let mut peer_state_lock = inner_state_lock.lock().unwrap();
 					peer_state_lock.prune_expired_request_state();
+					let removed_terminal_state =
+						peer_state_lock.prune_terminal_state_without_channels(&live_channel_ids);
 					let is_prunable = peer_state_lock.is_prunable();
 					if is_prunable {
-						need_remove.push(*counterparty_node_id);
+						need_remove.push((*counterparty_node_id, removed_terminal_state));
 					} else if peer_state_lock.needs_persist {
-						need_persist.push(*counterparty_node_id);
+						need_persist.push((*counterparty_node_id, removed_terminal_state));
 					}
 				}
 			}
 
-			for counterparty_node_id in need_persist.into_iter() {
-				debug_assert!(!need_remove.contains(&counterparty_node_id));
+			let has_removed_terminal_state = need_persist
+				.iter()
+				.chain(need_remove.iter())
+				.any(|(_, removed_terminal_state)| !removed_terminal_state.is_empty());
+			if has_removed_terminal_state {
+				// The peer state changes above are persisted through `need_persist` and
+				// `need_remove`; keep these in-memory lookup indexes in sync with them.
+				let mut peer_by_intercept_scid = self.peer_by_intercept_scid.write().unwrap();
+				let mut peer_by_channel_id = self.peer_by_channel_id.write().unwrap();
+				for (_, removed_terminal_state) in need_persist.iter().chain(need_remove.iter()) {
+					for (intercept_scid, channel_id) in removed_terminal_state.iter() {
+						peer_by_intercept_scid.remove(intercept_scid);
+						peer_by_channel_id.remove(channel_id);
+					}
+				}
+			}
+
+			for (counterparty_node_id, _) in need_persist.into_iter() {
+				debug_assert!(!need_remove
+					.iter()
+					.any(|(node_id, _)| node_id == &counterparty_node_id));
 				self.persist_peer_state(counterparty_node_id).await?;
 				did_persist = true;
 			}
 
-			for counterparty_node_id in need_remove {
+			for (counterparty_node_id, _) in need_remove {
 				let mut future_opt = None;
 				{
 					// We need to take the `per_peer_state` write lock to remove an entry, but also
@@ -2812,6 +2867,68 @@ mod tests {
 		assert_eq!(fee_payment.htlcs, vec![htlc]);
 	}
 
+	#[test]
+	fn prunes_terminal_state_for_closed_channels() {
+		let opening_fee_params = LSPS2OpeningFeeParams {
+			min_fee_msat: 10_000_000,
+			proportional: 10_000,
+			valid_until: LSPSDateTime::from_str("2035-05-20T08:30:45Z").unwrap(),
+			min_lifetime: 4032,
+			max_client_to_self_delay: 2016,
+			min_payment_size_msat: 10_000_000,
+			max_payment_size_msat: 1_000_000_000,
+			promise: "ignore".to_string(),
+		};
+		let stale_intercept_scid = 42;
+		let stale_user_channel_id = 43;
+		let stale_channel_id = ChannelId([44; 32]);
+		let live_intercept_scid = 45;
+		let live_user_channel_id = 46;
+		let live_channel_id = ChannelId([47; 32]);
+
+		let mut stale_jit_channel =
+			OutboundJITChannel::new(None, opening_fee_params.clone(), stale_user_channel_id, false);
+		stale_jit_channel.state =
+			OutboundJITChannelState::PaymentForwarded { channel_id: stale_channel_id };
+		let mut live_jit_channel =
+			OutboundJITChannel::new(None, opening_fee_params, live_user_channel_id, false);
+		live_jit_channel.state =
+			OutboundJITChannelState::PaymentForwarded { channel_id: live_channel_id };
+
+		let mut peer_state = PeerState::new();
+		peer_state.insert_outbound_channel(stale_intercept_scid, stale_jit_channel);
+		peer_state.insert_outbound_channel(live_intercept_scid, live_jit_channel);
+		peer_state
+			.intercept_scid_by_user_channel_id
+			.insert(stale_user_channel_id, stale_intercept_scid);
+		peer_state
+			.intercept_scid_by_user_channel_id
+			.insert(live_user_channel_id, live_intercept_scid);
+		peer_state.intercept_scid_by_channel_id.insert(stale_channel_id, stale_intercept_scid);
+		peer_state.intercept_scid_by_channel_id.insert(live_channel_id, live_intercept_scid);
+		peer_state.needs_persist = false;
+
+		assert_eq!(
+			peer_state.prune_terminal_state_without_channels(&[live_channel_id]),
+			vec![(stale_intercept_scid, stale_channel_id)]
+		);
+		assert!(!peer_state
+			.outbound_channels_by_intercept_scid
+			.contains_key(&stale_intercept_scid));
+		assert!(peer_state.outbound_channels_by_intercept_scid.contains_key(&live_intercept_scid));
+		assert!(!peer_state.intercept_scid_by_user_channel_id.contains_key(&stale_user_channel_id));
+		assert_eq!(
+			peer_state.intercept_scid_by_user_channel_id.get(&live_user_channel_id),
+			Some(&live_intercept_scid)
+		);
+		assert!(!peer_state.intercept_scid_by_channel_id.contains_key(&stale_channel_id));
+		assert_eq!(
+			peer_state.intercept_scid_by_channel_id.get(&live_channel_id),
+			Some(&live_intercept_scid)
+		);
+		assert!(peer_state.needs_persist);
+	}
+
 	#[test]
 	fn broadcast_not_allowed_after_non_paying_fee_payment_claimed() {
 		let min_fee_msat: u64 = 12345;