Send held_htlc_available with counterparty reply path

diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
index b3cac74..efdb516 100644
--- a/lightning/src/ln/channel.rs
+++ b/lightning/src/ln/channel.rs
@@ -70,6 +70,7 @@ use crate::ln::onion_utils::{
 use crate::ln::script::{self, ShutdownScript};
 use crate::ln::types::ChannelId;
 use crate::ln::LN_MAX_MSG_LEN;
+use crate::offers::static_invoice::StaticInvoice;
 use crate::routing::gossip::NodeId;
 use crate::sign::ecdsa::EcdsaChannelSigner;
 use crate::sign::tx_builder::{HTLCAmountDirection, NextCommitmentStats, SpecTxBuilder, TxBuilder};
@@ -8184,10 +8185,25 @@ where
 	/// waiting on this revoke_and_ack. The generation of this new commitment_signed may also fail,
 	/// generating an appropriate error *after* the channel state has been updated based on the
 	/// revoke_and_ack message.
+	///
+	/// The static invoices will be used by us as an async sender to enqueue [`HeldHtlcAvailable`]
+	/// onion messages for the often-offline recipient, and the blinded reply paths the invoices are
+	/// paired with were created by our channel counterparty and will be used as reply paths for
+	/// corresponding [`ReleaseHeldHtlc`] messages.
+	///
+	/// [`HeldHtlcAvailable`]: crate::onion_message::async_payments::HeldHtlcAvailable
+	/// [`ReleaseHeldHtlc`]: crate::onion_message::async_payments::ReleaseHeldHtlc
 	pub fn revoke_and_ack<F: Deref, L: Deref>(
 		&mut self, msg: &msgs::RevokeAndACK, fee_estimator: &LowerBoundedFeeEstimator<F>,
 		logger: &L, hold_mon_update: bool,
-	) -> Result<(Vec<(HTLCSource, PaymentHash)>, Option<ChannelMonitorUpdate>), ChannelError>
+	) -> Result<
+		(
+			Vec<(HTLCSource, PaymentHash)>,
+			Vec<(StaticInvoice, BlindedMessagePath)>,
+			Option<ChannelMonitorUpdate>,
+		),
+		ChannelError,
+	>
 	where
 		F::Target: FeeEstimator,
 		L::Target: Logger,
@@ -8302,6 +8318,7 @@ where
 		let mut finalized_claimed_htlcs = Vec::new();
 		let mut update_fail_htlcs = Vec::new();
 		let mut update_fail_malformed_htlcs = Vec::new();
+		let mut static_invoices = Vec::new();
 		let mut require_commitment = false;
 		let mut value_to_self_msat_diff: i64 = 0;
 
@@ -8417,6 +8434,24 @@ where
 				}
 			}
 			for htlc in pending_outbound_htlcs.iter_mut() {
+				for (htlc_id, blinded_path) in &msg.release_htlc_message_paths {
+					if htlc.htlc_id != *htlc_id {
+						continue;
+					}
+					let static_invoice = match htlc.source.static_invoice() {
+						Some(inv) if htlc.hold_htlc.is_some() => inv,
+						_ => {
+							// We should only be using our counterparty's release_htlc_message_path if we
+							// originally configured the HTLC to be held with them until the recipient comes
+							// online. Otherwise, our counterparty could include paths for all of our HTLCs and
+							// use the responses sent to their paths to determine which of our HTLCs are async
+							// payments.
+							log_trace!(logger, "Counterparty included release_htlc_message_path for non-async payment HTLC {}", htlc_id);
+							continue;
+						},
+					};
+					static_invoices.push((static_invoice, blinded_path.clone()));
+				}
 				if let OutboundHTLCState::LocalAnnounced(_) = htlc.state {
 					log_trace!(
 						logger,
@@ -8484,9 +8519,9 @@ where
 					self.context
 						.blocked_monitor_updates
 						.push(PendingChannelMonitorUpdate { update: monitor_update });
-					return Ok(($htlcs_to_fail, None));
+					return Ok(($htlcs_to_fail, static_invoices, None));
 				} else {
-					return Ok(($htlcs_to_fail, Some(monitor_update)));
+					return Ok(($htlcs_to_fail, static_invoices, Some(monitor_update)));
 				}
 			};
 		}
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 523de99..c1d38a0 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -873,6 +873,16 @@ impl HTLCSource {
 			_ => None,
 		}
 	}
+
+	pub(crate) fn static_invoice(&self) -> Option<StaticInvoice> {
+		match self {
+			Self::OutboundRoute {
+				bolt12_invoice: Some(PaidBolt12Invoice::StaticInvoice(inv)),
+				..
+			} => Some(inv.clone()),
+			_ => None,
+		}
+	}
 }
 
 /// This enum is used to specify which error data to send to peers when failing back an HTLC
@@ -11011,7 +11021,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 
 	#[rustfmt::skip]
 	fn internal_revoke_and_ack(&self, counterparty_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), MsgHandleErrInternal> {
-		let htlcs_to_fail = {
+		let (htlcs_to_fail, static_invoices) = {
 			let per_peer_state = self.per_peer_state.read().unwrap();
 			let mut peer_state_lock = per_peer_state.get(counterparty_node_id)
 				.ok_or_else(|| {
@@ -11027,7 +11037,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 						let mon_update_blocked = self.raa_monitor_updates_held(
 							&peer_state.actions_blocking_raa_monitor_updates, msg.channel_id,
 							*counterparty_node_id);
-						let (htlcs_to_fail, monitor_update_opt) = try_channel_entry!(self, peer_state,
+						let (htlcs_to_fail, static_invoices, monitor_update_opt) = try_channel_entry!(self, peer_state,
 							chan.revoke_and_ack(&msg, &self.fee_estimator, &&logger, mon_update_blocked), chan_entry);
 						if let Some(monitor_update) = monitor_update_opt {
 							let funding_txo = funding_txo_opt
@@ -11035,7 +11045,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 							handle_new_monitor_update!(self, funding_txo, monitor_update,
 								peer_state_lock, peer_state, per_peer_state, chan);
 						}
-						htlcs_to_fail
+						(htlcs_to_fail, static_invoices)
 					} else {
 						return try_channel_entry!(self, peer_state, Err(ChannelError::close(
 							"Got a revoke_and_ack message for an unfunded channel!".into())), chan_entry);
@@ -11045,6 +11055,10 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 			}
 		};
 		self.fail_holding_cell_htlcs(htlcs_to_fail, msg.channel_id, counterparty_node_id);
+		for (static_invoice, reply_path) in static_invoices {
+			let res = self.flow.enqueue_held_htlc_available(&static_invoice, HeldHtlcReplyPath::ToCounterparty { path: reply_path });
+			debug_assert!(res.is_ok(), "enqueue_held_htlc_available can only fail for non-async senders");
+		}
 		Ok(())
 	}