Introduce handle_message_received in ChannelMessageHandler

- Introduce the `handle_message_received` function to manage the
  behavior when a message is received from any peer.
- This function is used within `ChannelManager` to retry `InvoiceRequest`
  messages if we haven't received the corresponding invoice yet.
- This change makes the offer communication more robust against sudden
  connection drops where the initial attempt to send the message might
  have failed.

Author: shaavan

lightning-net-tokio/src/lib.rs

@@ -787,6 +787,7 @@ mod tests {
 		fn get_chain_hashes(&self) -> Option<Vec<ChainHash>> {
 			Some(vec![ChainHash::using_genesis_block(Network::Testnet)])
 		}
+		fn handle_message_received(&self) {}
 	}
 	impl MessageSendEventsProvider for MsgHandler {
 		fn get_and_clear_pending_msg_events(&self) -> Vec<MessageSendEvent> {

lightning/src/ln/channelmanager.rs

@@ -10746,6 +10746,39 @@ where
 			"Dual-funded channels not supported".to_owned(),
 			 msg.channel_id.clone())), *counterparty_node_id);
 	}
+
+	fn handle_message_received(&self) {
+		for (payment_id, retryable_invoice_request) in self
+			.pending_outbound_payments
+			.release_invoice_requests_awaiting_invoice()
+		{
+			let RetryableInvoiceRequest { invoice_request, nonce } = retryable_invoice_request;
+			let hmac = payment_id.hmac_for_offer_payment(nonce, &self.inbound_payment_key);
+			let context = OffersContext::OutboundPayment {
+				payment_id,
+				nonce,
+				hmac: Some(hmac)
+			};
+			match self.create_blinded_paths(context) {
+				Ok(reply_paths) => match self.enqueue_invoice_request(invoice_request, reply_paths) {
+					Ok(_) => {}
+					Err(_) => {
+						log_warn!(self.logger,
+							"Retry failed for an invoice request with payment_id: {}",
+							payment_id
+						);
+					}
+				},
+				Err(_) => {
+					log_warn!(self.logger,
+						"Retry failed for an invoice request with payment_id: {}. \
+							Reason: router could not find a blinded path to include as the reply path",
+						payment_id
+					);
+				}
+			}
+		}
+	}
 }
 
 impl<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, L: Deref>

lightning/src/ln/msgs.rs

@@ -1605,6 +1605,10 @@ pub trait ChannelMessageHandler : MessageSendEventsProvider {
 	/// If it's `None`, then no particular network chain hash compatibility will be enforced when
 	/// connecting to peers.
 	fn get_chain_hashes(&self) -> Option<Vec<ChainHash>>;
+
+	/// Handles behavior which is triggered when any message from any peer for any handler
+	/// is received.
+	fn handle_message_received(&self);
 }
 
 /// A trait to describe an object which can receive routing messages.

lightning/src/ln/peer_handler.rs

@@ -387,6 +387,8 @@ impl ChannelMessageHandler for ErroringMessageHandler {
 	fn handle_tx_abort(&self, their_node_id: &PublicKey, msg: &msgs::TxAbort) {
 		ErroringMessageHandler::push_error(self, their_node_id, msg.channel_id);
 	}
+
+	fn handle_message_received(&self) {}
 }
 
 impl Deref for ErroringMessageHandler {
@@ -1750,6 +1752,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 			log_trace!(logger, "Received message {:?} from {}", message, log_pubkey!(their_node_id));
 		}
 
+		self.message_handler.chan_handler.handle_message_received();
+
 		let mut should_forward = None;
 
 		match message {

lightning/src/util/test_utils.rs

@@ -936,6 +936,8 @@ impl msgs::ChannelMessageHandler for TestChannelMessageHandler {
 	fn handle_tx_abort(&self, _their_node_id: &PublicKey, msg: &msgs::TxAbort) {
 		self.received_msg(wire::Message::TxAbort(msg.clone()));
 	}
+
+	fn handle_message_received(&self) {}
 }
 
 impl events::MessageSendEventsProvider for TestChannelMessageHandler {