Narrow TOCTOU window in execute_htlc_actions

forward_intercepted_htlc consumes the InterceptId on Ok, but Ok
only means the HTLC was queued internally — the actual send_htlc
can still fail if the peer disconnects before the commitment
update is exchanged. This created two problems:

1. The peer could disconnect between computing actions (usability
   check passes) and executing the forward. The forward would
   succeed at the LDK API layer but silently fail at send_htlc,
   leaving the HTLC in limbo with a consumed InterceptId.

2. Removing the HTLC from the store immediately after Ok meant
   the retry timer could never re-attempt it. If the internal
   send_htlc failed, the HTLC was lost until a new
   HTLCIntercepted event arrived with a fresh InterceptId.

Add a pre-forward peer liveness check to skip forwarding when
the peer has already disconnected. On Ok, keep the HTLC in the
store so the retry timer can detect and re-process it. On Err,
remove it — the InterceptId is consumed or stale, so a new
HTLCIntercepted event will arrive for a fresh retry.

An alternative considered was making forward_intercepted_htlc
itself check peer liveness, but that would require plumbing
connection state into the channel manager's intercept path. The
pre-check here is simpler and sufficient — the residual race
window is sub-millisecond.

Author: amackillop

lightning-liquidity/src/lsps4/service.rs

@@ -726,29 +726,17 @@ where
 	) {
 		// Execute forwards
 		for forward_action in actions.forwards {
-			// Log channel state RIGHT BEFORE forwarding - key diagnostic
-			let channels = self
-				.channel_manager
-				.get_cm()
-				.list_channels_with_counterparty(&their_node_id);
-			let target_channel =
-				channels.iter().find(|ch| ch.channel_id == forward_action.channel_id);
-			if let Some(ch) = target_channel {
+			// Re-check peer liveness right before forwarding to narrow the
+			// TOCTOU window between the usability check and the actual forward.
+			if !self.is_peer_connected(&their_node_id) {
 				log_info!(
 					self.logger,
-					"[LSPS4] execute_htlc_actions: PRE-FORWARD channel {} - is_usable: {}, is_channel_ready: {}, outbound_capacity: {}msat",
-					ch.channel_id,
-					ch.is_usable,
-					ch.is_channel_ready,
-					ch.outbound_capacity_msat
-				);
-			} else {
-				log_error!(
-					self.logger,
-					"[LSPS4] execute_htlc_actions: TARGET CHANNEL {} NOT FOUND for peer {} before forward!",
-					forward_action.channel_id,
-					their_node_id
+					"[LSPS4] execute_htlc_actions: peer {} disconnected before forward, 					 skipping HTLC {:?} (will retry on next timer tick). payment_hash: {}",
+					their_node_id,
+					forward_action.htlc.id(),
+					forward_action.htlc.payment_hash()
 				);
+				continue;
 			}
 
 			log_info!(
@@ -769,40 +757,46 @@ where
 				forward_action.amount_to_forward_msat,
 			) {
 				Ok(()) => {
+					// Ok means the InterceptId was consumed and the HTLC queued for
+					// forwarding. However, the internal send_htlc can still fail if the
+					// peer disconnects between now and the commitment update. Keep the
+					// HTLC in the store; the next timer tick will attempt to re-forward
+					// the (now consumed) InterceptId, get Err, and clean up. If the
+					// internal forward truly failed, a new HTLCIntercepted event will
+					// arrive with a fresh InterceptId and be processed normally.
 					log_info!(
 						self.logger,
-						"[LSPS4] execute_htlc_actions: forward_intercepted_htlc returned Ok for HTLC {:?} (took {}ms). NOTE: Ok does NOT guarantee delivery - channel may still be reestablishing.",
+						"[LSPS4] execute_htlc_actions: forward_intercepted_htlc returned Ok for HTLC {:?} 						 (took {}ms), keeping in store until confirmed. payment_hash: {}",
 						forward_action.htlc.id(),
-						fwd_start.elapsed().as_millis()
+						fwd_start.elapsed().as_millis(),
+						forward_action.htlc.payment_hash()
 					);
 				},
 				Err(ref e) => {
-					log_error!(
-						self.logger,
-						"[LSPS4] execute_htlc_actions: forward_intercepted_htlc FAILED for HTLC {:?} - error: {:?} (took {}ms)",
-						forward_action.htlc.id(),
-						e,
-						fwd_start.elapsed().as_millis()
-					);
-				},
-			}
-
-			// Remove from store - log whether it was actually present
-			match self.htlc_store.remove(&forward_action.htlc.id()) {
-				Ok(()) => {
-					log_info!(
-						self.logger,
-						"[LSPS4] execute_htlc_actions: removed HTLC {:?} from store after forward",
-						forward_action.htlc.id()
-					);
-				},
-				Err(e) => {
+					// InterceptId was already consumed or not found. Clean up the store.
 					log_info!(
 						self.logger,
-						"[LSPS4] execute_htlc_actions: HTLC {:?} was NOT in store (expected if from htlc_intercepted connected path): {}",
+						"[LSPS4] execute_htlc_actions: forward returned Err for HTLC {:?}: {:?} 						 (took {}ms), removing from store. payment_hash: {}",
 						forward_action.htlc.id(),
-						e
+						e,
+						fwd_start.elapsed().as_millis(),
+						forward_action.htlc.payment_hash()
 					);
+					match self.htlc_store.remove(&forward_action.htlc.id()) {
+						Ok(()) => log_info!(
+							self.logger,
+							"[LSPS4] execute_htlc_actions: removed HTLC {:?} from store. payment_hash: {}",
+							forward_action.htlc.id(),
+							forward_action.htlc.payment_hash()
+						),
+						Err(e) => log_info!(
+							self.logger,
+							"[LSPS4] execute_htlc_actions: HTLC {:?} not in store (already removed): {}. payment_hash: {}",
+							forward_action.htlc.id(),
+							e,
+							forward_action.htlc.payment_hash()
+						),
+					}
 				},
 			}
 		}