Track pending channel opens to prevent duplicates

After the previous commit moved usability checks to execute time,
the timer can call process_htlcs_for_peer repeatedly while a
channel is still opening. calculate_htlc_actions_for_peer sees no
is_channel_ready channels and requests a new one each time,
producing duplicate OpenChannel events.

Add a pending_channel_opens set (RwLock<HashSet<PublicKey>>).
execute_htlc_actions inserts the peer when it emits OpenChannel,
and channel_ready removes it. If the set already contains the
peer, the OpenChannel is suppressed.

calculate_htlc_actions_for_peer now filters by is_channel_ready
instead of including all channels. Channels still opening
(is_channel_ready=false) report outbound_capacity_msat but reject
forwards with "Channel is still opening", consuming the
InterceptId and losing the HTLC. These are zero-conf channels, so
on-chain confirmation is not the issue; the channel simply hasn't
finished its opening handshake yet. Reestablishing channels
(is_channel_ready=true, is_usable=false) can forward once
reestablish completes and are included, preserving the
spurious-open fix from the previous commit.

Author: amackillop

lightning-liquidity/src/lsps4/service.rs

@@ -101,6 +101,7 @@ where
 	scid_store: ScidStore<L, K>,
 	htlc_store: HTLCStore<L, K>,
 	connected_peers: RwLock<HashSet<PublicKey>>,
+	pending_channel_opens: RwLock<HashSet<PublicKey>>,
 	config: LSPS4ServiceConfig,
 }
 
@@ -128,6 +129,7 @@ where
 			config,
 			logger,
 			connected_peers: RwLock::new(HashSet::new()),
+			pending_channel_opens: RwLock::new(HashSet::new()),
 		})
 	}
 
@@ -257,6 +259,7 @@ where
 	pub fn channel_ready(
 		&self, counterparty_node_id: &PublicKey,
 	) -> Result<(), APIError> {
+		self.pending_channel_opens.write().unwrap().remove(counterparty_node_id);
 		let is_connected = self.is_peer_connected(counterparty_node_id);
 
 		log_info!(
@@ -581,8 +584,17 @@ where
 			self.channel_manager.get_cm().list_channels_with_counterparty(&their_node_id);
 		let channel_count = channels.len();
 
+		// Include channels that finished opening (is_channel_ready) in capacity
+		// decisions, even if still reestablishing (is_usable=false). This prevents
+		// spurious channel opens during reestablish. Channels still opening
+		// (is_channel_ready=false) are excluded: they report outbound_capacity_msat
+		// but reject forwards, consuming the InterceptId and losing the HTLC.
+		// execute_htlc_actions checks usability before forwarding.
 		let mut channel_capacity_map: HashMap<ChannelId, u64> = new_hash_map();
 		for channel in &channels {
+			if !channel.is_channel_ready {
+				continue;
+			}
 			channel_capacity_map.insert(channel.channel_id, channel.outbound_capacity_msat);
 			log_info!(
 				self.logger,
@@ -773,21 +785,30 @@ where
 			}
 		}
 
-		// Handle new channel opening
+		// Handle new channel opening — skip if one is already in flight.
 		if let Some(channel_size_msat) = actions.new_channel_needed_msat {
+			if self.pending_channel_opens.read().unwrap().contains(&their_node_id) {
+				log_info!(
+					self.logger,
+					"[LSPS4] execute_htlc_actions: peer {} needs a new channel but one is \
+					 already opening, skipping duplicate OpenChannel",
+					their_node_id
+				);
+			} else {
 				log_info!(
 					self.logger,
 					"Need a new channel with peer {} for {}msat to forward HTLCs",
 					their_node_id,
 					channel_size_msat
 				);
-
+				self.pending_channel_opens.write().unwrap().insert(their_node_id);
 				let mut event_queue_notifier = self.pending_events.notifier();
 				event_queue_notifier.enqueue(crate::events::LiquidityEvent::LSPS4Service(LSPS4ServiceEvent::OpenChannel {
 					their_network_key: their_node_id,
 					amt_to_forward_msat: channel_size_msat,
 					channel_count: actions.channel_count,
 				}));
+			}
 		}
 	}